fix(server): static 400s and opt-in identifier scrub (alt to #77)

server/src/api/preferences.rs

@@ -228,11 +228,10 @@ pub async fn write(
         if p.category.is_empty() || p.category.len() > 255 {
             return Err(ApiError::bad_request("category must be 1–255 characters"));
         }
+        // Static message. The submitted channel value is caller input and
+        // must not be reflected into the response body.
         if !ALLOWED_CHANNELS.contains(&p.channel.as_str()) {
-            return Err(ApiError::bad_request(format!(
-                "unknown channel: {}",
-                p.channel
-            )));
+            return Err(ApiError::bad_request("unknown channel"));
         }
     }
 

server/src/config.rs

@@ -78,6 +78,13 @@ pub struct Config {
     /// In-flight job drain budget after claiming stops. Past it the worker
     /// is aborted. At-least-once semantics make the rollback safe.
     pub shutdown_drain_deadline: Duration,
+    /// Opt-in identifier scrubbing for the access log and the http.request
+    /// span. When set, subscriber and environment identifiers (query params
+    /// and the subscriber id path segment, on the public and admin planes
+    /// both) are replaced with truncated SHA-256 hashes, for operators whose
+    /// logs leave their control. Credentials are scrubbed regardless of this
+    /// flag.
+    pub log_scrub_identifiers: bool,
 }
 
 /// Manual impl so credentials can never reach logs through `{:?}`.
@@ -130,6 +137,7 @@ impl std::fmt::Debug for Config {
             .field("subscriber_rate_burst", &self.subscriber_rate_burst)
             .field("shutdown_readiness_grace", &self.shutdown_readiness_grace)
             .field("shutdown_drain_deadline", &self.shutdown_drain_deadline)
+            .field("log_scrub_identifiers", &self.log_scrub_identifiers)
             .finish()
     }
 }
@@ -197,6 +205,7 @@ impl Config {
                 "CHIMELY_SHUTDOWN_DRAIN_DEADLINE_MS",
                 30_000,
             )?),
+            log_scrub_identifiers: parse_var("CHIMELY_LOG_SCRUB_IDENTIFIERS", false)?,
         })
     }
 }
@@ -249,6 +258,7 @@ mod tests {
             subscriber_rate_burst: 0.0,
             shutdown_readiness_grace: Duration::from_millis(1),
             shutdown_drain_deadline: Duration::from_millis(1),
+            log_scrub_identifiers: false,
         };
         let out = format!("{cfg:?}");
         assert!(
@@ -259,4 +269,22 @@ mod tests {
         assert!(out.contains("listen_addr"));
         assert!(out.contains("ops@example.com"));
     }
+
+    /// Identifier scrubbing is opt-in. An absent variable must parse to
+    /// false. nextest runs every test in its own process, so mutating this
+    /// process's environment cannot race another test.
+    #[test]
+    fn log_scrub_identifiers_env_var_defaults_off() {
+        unsafe {
+            std::env::set_var("DATABASE_URL", "postgres://localhost/chimely");
+            std::env::remove_var("CHIMELY_LOG_SCRUB_IDENTIFIERS");
+        }
+        assert!(!Config::from_env().unwrap().log_scrub_identifiers);
+
+        unsafe { std::env::set_var("CHIMELY_LOG_SCRUB_IDENTIFIERS", "true") }
+        assert!(Config::from_env().unwrap().log_scrub_identifiers);
+
+        unsafe { std::env::set_var("CHIMELY_LOG_SCRUB_IDENTIFIERS", "false") }
+        assert!(!Config::from_env().unwrap().log_scrub_identifiers);
+    }
 }

server/src/error.rs

@@ -101,6 +101,12 @@ impl IntoResponse for ApiError {
 
 /// Internal failures (database, serialization) become opaque 500s. The detail
 /// goes to tracing, never to the client.
+///
+/// The `?err` line below logs sqlx errors verbatim, and a Postgres unique
+/// violation carries `DETAIL: Key (...)=(...)` with the conflicting values.
+/// Any INSERT that can conflict on a secret-keyed or PII-keyed table must
+/// intercept the conflict before the error reaches this conversion, the way
+/// idempotency conflicts and admin email conflicts already do.
 impl<E> From<E> for ApiError
 where
     E: Into<anyhow::Error>,

server/src/extract.rs

@@ -4,6 +4,10 @@
 //! plain-text 400/415/422 bodies. The contract allows exactly one error
 //! shape (`{"error": {"code", "message"}}`) and no 415/422, so every handler
 //! extracts through these wrappers instead.
+//!
+//! Rejection messages are static. axum/serde rejection prose can quote a
+//! caller scalar on type mismatch, so no request-derived text may reach the
+//! response body.
 
 use axum::extract::{FromRequest, FromRequestParts, OptionalFromRequest, Request};
 use axum::http::request::Parts;
@@ -23,7 +27,7 @@ where
     async fn from_request(req: Request, state: &S) -> Result<Self, ApiError> {
         match <axum::Json<T> as FromRequest<S>>::from_request(req, state).await {
             Ok(axum::Json(value)) => Ok(Self(value)),
-            Err(rejection) => Err(ApiError::bad_request(rejection.body_text())),
+            Err(_) => Err(ApiError::bad_request("invalid request body")),
         }
     }
 }
@@ -39,7 +43,7 @@ where
         match <axum::Json<T> as OptionalFromRequest<S>>::from_request(req, state).await {
             Ok(Some(axum::Json(value))) => Ok(Some(Self(value))),
             Ok(None) => Ok(None),
-            Err(rejection) => Err(ApiError::bad_request(rejection.body_text())),
+            Err(_) => Err(ApiError::bad_request("invalid request body")),
         }
     }
 }
@@ -56,7 +60,7 @@ where
     async fn from_request_parts(parts: &mut Parts, state: &S) -> Result<Self, ApiError> {
         match axum::extract::Query::<T>::from_request_parts(parts, state).await {
             Ok(axum::extract::Query(value)) => Ok(Self(value)),
-            Err(rejection) => Err(ApiError::bad_request(rejection.body_text())),
+            Err(_) => Err(ApiError::bad_request("invalid query string")),
         }
     }
 }

server/src/http.rs

@@ -120,7 +120,7 @@ pub fn router(state: AppState) -> Router {
             get(move || std::future::ready(prometheus.render())),
         )
         .merge(utoipa_scalar::Scalar::with_url("/docs", openapi::api_doc()))
-        .layer(middleware::from_fn(access_log))
+        .layer(middleware::from_fn_with_state(state.clone(), access_log))
         .with_state(state)
 }
 
@@ -244,18 +244,24 @@ async fn readyz(
 /// credential on the SSE endpoint (EventSource cannot set headers) and must
 /// never reach log lines. This is a tested invariant.
 ///
+/// Identifier scrubbing (`log_scrub_identifiers`) applies to the log line
+/// and the `http.request` span both, so scrubbed identifiers cannot leak
+/// through OTLP export either.
+///
 /// The handler runs inside an `http.request` span: jobs enqueued by the
 /// handler carry this span's context through the outbox, so the whole
 /// ingest -> outbox -> worker -> hint flow lands in one trace.
 async fn access_log(
+    axum::extract::State(state): axum::extract::State<AppState>,
     req: axum::extract::Request,
     next: middleware::Next,
 ) -> axum::response::Response {
     use tracing::Instrument as _;
 
+    let scrub_identifiers = state.cfg.log_scrub_identifiers;
     let method = req.method().clone();
-    let path = req.uri().path().to_owned();
-    let query = req.uri().query().map(scrub_query);
+    let path = scrub_path(req.uri().path(), scrub_identifiers);
+    let query = req.uri().query().map(|q| scrub_query(q, scrub_identifiers));
     let started = std::time::Instant::now();
     let span = tracing::info_span!("http.request", %method, %path);
     let response = next.run(req).instrument(span).await;
@@ -276,11 +282,20 @@ async fn access_log(
 /// valid credential and must scrub the same way) and the output is
 /// re-encoded, so a credential value can never smuggle raw bytes into a log
 /// line.
-pub fn scrub_query(query: &str) -> String {
+///
+/// With `scrub_identifiers` set, `subscriber_id` and `environment` values
+/// are replaced with truncated hashes. The credential scrub does not depend
+/// on the flag.
+pub fn scrub_query(query: &str, scrub_identifiers: bool) -> String {
     form_urlencoded::parse(query.as_bytes())
         .map(|(name, value)| {
             let value = if name.eq_ignore_ascii_case("subscriber_hash") {
                 std::borrow::Cow::Borrowed("redacted")
+            } else if scrub_identifiers
+                && (name.eq_ignore_ascii_case("subscriber_id")
+                    || name.eq_ignore_ascii_case("environment"))
+            {
+                std::borrow::Cow::Owned(hash_identifier(&value))
             } else {
                 value
             };
@@ -294,31 +309,128 @@ pub fn scrub_query(query: &str) -> String {
         .join("&")
 }
 
+/// Replaces the path segment that follows a `subscribers` segment with a
+/// truncated hash when identifier scrubbing is enabled. The positional rule
+/// covers `/v1/subscribers/{id}` and the admin plane's
+/// `/admin/api/environments/{env_id}/subscribers/{subscriber_id}` alike.
+/// Other paths pass through unchanged.
+pub fn scrub_path(path: &str, scrub_identifiers: bool) -> String {
+    if !scrub_identifiers {
+        return path.to_owned();
+    }
+    let mut previous = "";
+    path.split('/')
+        .map(|segment| {
+            let scrubbed = if previous == "subscribers" && !segment.is_empty() {
+                std::borrow::Cow::Owned(hash_identifier(segment))
+            } else {
+                std::borrow::Cow::Borrowed(segment)
+            };
+            previous = segment;
+            scrubbed
+        })
+        .collect::<Vec<_>>()
+        .join("/")
+}
+
+/// Truncated SHA-256 of an identifier for scrubbed log output. Stable per
+/// input, so operators can still correlate log lines by hashing a known
+/// identifier offline.
+fn hash_identifier(value: &str) -> String {
+    use sha2::Digest as _;
+    hex::encode(&sha2::Sha256::digest(value.as_bytes())[..6])
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
 
+    // Expected hashes are the first 12 hex chars of SHA-256, computed
+    // outside this codebase (shasum -a 256).
+    const HASH_USR_123: &str = "ca010ec7feb3";
+    const HASH_ACME: &str = "822b33ad87c1";
+
     #[test]
     fn scrubs_subscriber_hash_wherever_it_appears() {
         assert_eq!(
-            scrub_query("environment=acme&subscriber_id=u1&subscriber_hash=deadbeef"),
+            scrub_query(
+                "environment=acme&subscriber_id=u1&subscriber_hash=deadbeef",
+                false
+            ),
             "environment=acme&subscriber_id=u1&subscriber_hash=redacted"
         );
-        assert_eq!(scrub_query("subscriber_hash=x"), "subscriber_hash=redacted");
         assert_eq!(
-            scrub_query("SUBSCRIBER_HASH=x&a=b"),
+            scrub_query("subscriber_hash=x", false),
+            "subscriber_hash=redacted"
+        );
+        assert_eq!(
+            scrub_query("SUBSCRIBER_HASH=x&a=b", false),
             "SUBSCRIBER_HASH=redacted&a=b"
         );
         // Percent-encoded names decode before matching: auth accepts
         // `subscriber%5Fhash`, so the scrub must catch it too.
         assert_eq!(
-            scrub_query("subscriber%5Fhash=deadbeef"),
+            scrub_query("subscriber%5Fhash=deadbeef", false),
             "subscriber_hash=redacted"
         );
         assert_eq!(
-            scrub_query("%73ubscriber_hash=deadbeef&a=b"),
+            scrub_query("%73ubscriber_hash=deadbeef&a=b", false),
             "subscriber_hash=redacted&a=b"
         );
-        assert_eq!(scrub_query("a=b"), "a=b");
+        assert_eq!(scrub_query("a=b", false), "a=b");
+    }
+
+    #[test]
+    fn hashes_identifier_params_when_scrub_enabled() {
+        assert_eq!(
+            scrub_query(
+                "environment=acme&subscriber_id=usr_123&subscriber_hash=deadbeef",
+                true
+            ),
+            format!(
+                "environment={HASH_ACME}&subscriber_id={HASH_USR_123}&subscriber_hash=redacted"
+            )
+        );
+        // Name matching is case-insensitive and percent-decoded, like the
+        // credential scrub.
+        assert_eq!(
+            scrub_query("SUBSCRIBER_ID=usr_123", true),
+            format!("SUBSCRIBER_ID={HASH_USR_123}")
+        );
+        assert_eq!(
+            scrub_query("subscriber%5Fid=usr_123", true),
+            format!("subscriber_id={HASH_USR_123}")
+        );
+        assert_eq!(scrub_query("limit=10&a=b", true), "limit=10&a=b");
+    }
+
+    #[test]
+    fn hashes_admin_subscriber_path_segment_when_scrub_enabled() {
+        assert_eq!(
+            scrub_path("/admin/api/environments/env_1/subscribers/usr_123", true),
+            format!("/admin/api/environments/env_1/subscribers/{HASH_USR_123}")
+        );
+        assert_eq!(
+            scrub_path("/admin/api/environments/env_1/subscribers/usr_123", false),
+            "/admin/api/environments/env_1/subscribers/usr_123"
+        );
+    }
+
+    #[test]
+    fn hashes_subscriber_path_segment_when_scrub_enabled() {
+        assert_eq!(
+            scrub_path("/v1/subscribers/usr_123", true),
+            format!("/v1/subscribers/{HASH_USR_123}")
+        );
+        assert_eq!(
+            scrub_path("/v1/subscribers/usr_123/preferences", true),
+            format!("/v1/subscribers/{HASH_USR_123}/preferences")
+        );
+        assert_eq!(scrub_path("/v1/inbox/items", true), "/v1/inbox/items");
+        assert_eq!(scrub_path("/v1/subscribers/", true), "/v1/subscribers/");
+        assert_eq!(
+            scrub_path("/v1/subscribers/usr_123", false),
+            "/v1/subscribers/usr_123"
+        );
     }
 }

server/tests/chaos.rs

@@ -527,6 +527,7 @@ async fn sustained_jobs_churn_stays_bounded_under_autovacuum() {
         subscriber_rate_burst: 0.0,
         shutdown_readiness_grace: Duration::from_millis(100),
         shutdown_drain_deadline: Duration::from_secs(5),
+        log_scrub_identifiers: false,
     };
     let cfg = std::sync::Arc::new(cfg);
     let pubsub = chimely::pubsub::build(None, &pool).await.unwrap();

server/tests/inbox.rs

@@ -488,7 +488,9 @@ async fn muted_categories_disappear_from_the_list_at_read_time() {
     assert_eq!(body["preferences"].as_array().unwrap().len(), 0);
     assert_eq!(app.list_all_items(SUB, 10).await.len(), 3);
 
-    // Unknown channels are an API-layer 400 (no DB CHECK by design).
+    // Unknown channels are an API-layer 400 (no DB CHECK by design). The
+    // message is static. Echoing the caller's channel value would reflect
+    // request bytes into the body (H4, #58).
     let res = app
         .client
         .put(format!("{}/v1/inbox/preferences", app.base))
@@ -498,6 +500,15 @@ async fn muted_categories_disappear_from_the_list_at_read_time() {
         .await
         .unwrap();
     assert_eq!(res.status(), 400);
+    let body: serde_json::Value = res.json().await.unwrap();
+    assert_eq!(body["error"]["message"], "unknown channel");
+    assert!(
+        !body["error"]["message"]
+            .as_str()
+            .unwrap()
+            .contains("web_push"),
+        "rejection echoed caller input: {body}"
+    );
 }
 
 #[tokio::test]