Merge PR #2125 (aepod) into release integration

# Conflicts:
#	v2/crates/wifi-densepose-sensing-server/src/cli.rs
#	v2/crates/wifi-densepose-sensing-server/src/main.rs
This commit is contained in:
ruv committed 2026-10-05 11:45:26 -04:00
commit 678f63f4f3
8 files changed
+543 -27

No files matched your search

+2 -1
View File
@@ -128,7 +128,7 @@ Per ADR-115 §3.11.1, the Matter Bridge exposes a subset on standard clusters so
| `--mqtt-rate-rssi <HZ>` | 0.1 | RSSI publish rate (Hz) |
| `--mqtt-publish-pose` | off | Enable pose-keypoint publication |
| `--mqtt-rate-pose <HZ>` | 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 <PATH>` | — | 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.
+8 -1
View File
@@ -890,7 +890,14 @@ Full design + operator guide: [`docs/integrations/home-assistant.md`](integratio
sensing-server --mqtt --mqtt-host <broker> --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)
@@ -61,14 +61,18 @@ use wifi_densepose_sensing_server::mqtt::{
async fn main() -> Result<(), Box<dyn std::error::Error>> {
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<dyn std::error::Error>> {
// 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() => {
@@ -21,10 +21,11 @@ pub fn parse_env_bool(value: &str) -> Result<bool, String> {
}
}
/// 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
@@ -98,10 +99,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", value_parser = parse_env_bool)]
pub privacy_mode: bool,
}
/// CLI arguments for the sensing server.
@@ -282,11 +279,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", value_parser = parse_env_bool)]
pub privacy_mode: bool,
@@ -29,6 +29,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;
@@ -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};
@@ -180,7 +181,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", value_parser = cli::parse_env_bool)]
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`.
@@ -2000,6 +2008,9 @@ struct AppStateInner {
recording_current_id: Option<String>,
/// Shutdown signal for the recording writer task.
recording_stop_tx: Option<tokio::sync::watch::Sender<bool>>,
/// `--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`.
@@ -2180,6 +2191,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>) -> 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);
@@ -2707,6 +2894,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::<String>(256).0,
adaptive_model: None,
@@ -5378,9 +5566,9 @@ async fn ws_sensing_handler(
}
async fn handle_ws_client(mut socket: WebSocket, state: SharedState) {
let (mut rx, _client) = {
let (mut rx, privacy_mode, _client) = {
let s = state.read().await;
(s.tx.subscribe(), s.ws_clients.enter())
(s.tx.subscribe(), s.privacy_mode, s.ws_clients.enter())
};
info!("WebSocket client connected (sensing)");
@@ -5394,6 +5582,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;
}
@@ -5487,9 +5683,9 @@ async fn ws_pose_handler(
}
async fn handle_ws_pose_client(mut socket: WebSocket, state: SharedState) {
let (mut rx, _client) = {
let (mut rx, privacy_mode, _client) = {
let s = state.read().await;
(s.tx.subscribe(), s.ws_clients.enter())
(s.tx.subscribe(), s.privacy_mode, s.ws_clients.enter())
};
info!("WebSocket client connected (pose)");
@@ -5514,7 +5710,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",
@@ -5572,7 +5771,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,
@@ -5597,6 +5796,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;
}
@@ -7352,6 +7554,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 => {
@@ -7850,6 +8063,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!({
@@ -7873,6 +8088,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;
@@ -13424,6 +13647,13 @@ async fn main() {
// clients drop oldest, identical backpressure shape.
let (intro_tx, _) = broadcast::channel::<String>(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
@@ -13434,7 +13664,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();
@@ -13549,6 +13781,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::<String>(256).0,
@@ -13850,6 +14083,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
@@ -14022,6 +14259,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
@@ -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,
}
}
@@ -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<String> {
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<PrivacyFilter>,
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"));
}
}