diff --git a/server/src/api/preferences.rs b/server/src/api/preferences.rs index 4e3274e..86ab81f 100644 --- a/server/src/api/preferences.rs +++ b/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")); } } diff --git a/server/src/config.rs b/server/src/config.rs index 9a1ecea..3c7e270 100644 --- a/server/src/config.rs +++ b/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); + } } diff --git a/server/src/error.rs b/server/src/error.rs index 8a840b6..715dda4 100644 --- a/server/src/error.rs +++ b/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 From for ApiError where E: Into, diff --git a/server/src/extract.rs b/server/src/extract.rs index 553274a..b077114 100644 --- a/server/src/extract.rs +++ b/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 { match as FromRequest>::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 as OptionalFromRequest>::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 { match axum::extract::Query::::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")), } } } diff --git a/server/src/http.rs b/server/src/http.rs index 741a4f9..f72d47e 100644 --- a/server/src/http.rs +++ b/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, 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::>() + .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" + ); } } diff --git a/server/tests/chaos.rs b/server/tests/chaos.rs index 91dd700..d442e65 100644 --- a/server/tests/chaos.rs +++ b/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(); diff --git a/server/tests/inbox.rs b/server/tests/inbox.rs index b4eb068..93da268 100644 --- a/server/tests/inbox.rs +++ b/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] diff --git a/server/tests/log_scrub.rs b/server/tests/log_scrub.rs new file mode 100644 index 0000000..10d5c26 --- /dev/null +++ b/server/tests/log_scrub.rs @@ -0,0 +1,159 @@ +//! Identifier scrubbing across the config -> middleware -> output boundary. +//! A router configured with `log_scrub_identifiers` serves real requests and +//! the tests assert on captured access-log lines and `http.request` span +//! fields. nextest gives each test its own process, so the global subscriber +//! install below cannot leak elsewhere. + +mod support; + +use std::sync::{Arc, Mutex}; +use std::time::Duration; + +use sha2::Digest as _; +use support::SseStream; +use tracing_subscriber::fmt::format::FmtSpan; + +/// First 12 hex chars of SHA-256("usr_123"), computed outside this codebase +/// (shasum -a 256). +const HASH_USR_123: &str = "ca010ec7feb3"; + +/// Captures all log output for the process. Span-open events are enabled so +/// the `http.request` span's fields appear in the capture alongside the +/// access-log events. +fn install_log_capture() -> Arc>> { + struct BufWriter(Arc>>); + impl std::io::Write for BufWriter { + fn write(&mut self, buf: &[u8]) -> std::io::Result { + self.0.lock().unwrap().extend_from_slice(buf); + Ok(buf.len()) + } + fn flush(&mut self) -> std::io::Result<()> { + Ok(()) + } + } + let logs = Arc::new(Mutex::new(Vec::new())); + let sink = logs.clone(); + tracing_subscriber::fmt() + .with_max_level(tracing::Level::INFO) + .with_span_events(FmtSpan::NEW) + .with_ansi(false) + .with_writer(move || BufWriter(sink.clone())) + .try_init() + .expect("install capture subscriber"); + logs +} + +/// Drives one request carrying the subscriber id in the path and one +/// carrying subscriber and environment identifiers in the query string, +/// then returns the captured log output. +async fn capture_identifier_requests(app: &support::TestApp) -> String { + let logs = install_log_capture(); + + let res = app + .client + .put(format!("{}/v1/subscribers/usr_123", app.base)) + .bearer_auth(&app.env.api_key) + .send() + .await + .expect("upsert request"); + assert_eq!(res.status(), 200); + + let mut stream = SseStream::connect(app, "usr_123", None).await; + stream.next_frame(Duration::from_secs(2)).await; + drop(stream); + tokio::time::sleep(Duration::from_millis(200)).await; + + String::from_utf8_lossy(&logs.lock().unwrap()).to_string() +} + +fn find_line<'a>(captured: &'a str, needles: &[&str]) -> &'a str { + captured + .lines() + .find(|line| needles.iter().all(|needle| line.contains(needle))) + .unwrap_or_else(|| panic!("no log line containing {needles:?}:\n{captured}")) +} + +#[tokio::test] +async fn enabled_scrub_hashes_identifiers_in_access_log_and_span() { + let app = support::spawn_configured(false, |cfg| cfg.log_scrub_identifiers = true).await; + let captured = capture_identifier_requests(&app).await; + let hashed_path = format!("/v1/subscribers/{HASH_USR_123}"); + + let span_line = find_line(&captured, &["http.request{", "/v1/subscribers/"]); + assert!( + span_line.contains(&hashed_path), + "span path is not hashed: {span_line}" + ); + + let access_line = find_line(&captured, &["chimely::access", "/v1/subscribers/"]); + assert!( + access_line.contains(&hashed_path), + "access-log path is not hashed: {access_line}" + ); + assert!( + !captured.contains("/v1/subscribers/usr_123"), + "raw subscriber path leaked:\n{captured}" + ); + + let stream_line = find_line(&captured, &["chimely::access", "/v1/inbox/stream"]); + let env_hash = hex::encode(&sha2::Sha256::digest(app.env.slug.as_bytes())[..6]); + assert!( + stream_line.contains(&format!("subscriber_id={HASH_USR_123}")), + "subscriber_id is not hashed: {stream_line}" + ); + assert!( + stream_line.contains(&format!("environment={env_hash}")), + "environment is not hashed: {stream_line}" + ); + assert!( + stream_line.contains("subscriber_hash=redacted"), + "{stream_line}" + ); + assert!( + !stream_line.contains("usr_123"), + "raw subscriber id leaked: {stream_line}" + ); + assert!( + !stream_line.contains(&app.env.slug), + "raw environment leaked: {stream_line}" + ); +} + +/// The default-off contract. With the flag unset, identifiers log raw in +/// the access log and the span, exactly as before the flag existed. The +/// credential scrub stays on. +#[tokio::test] +async fn default_off_logs_raw_identifiers_unchanged() { + let app = support::spawn().await; + let captured = capture_identifier_requests(&app).await; + + let span_line = find_line(&captured, &["http.request{", "/v1/subscribers/"]); + assert!( + span_line.contains("/v1/subscribers/usr_123"), + "span path changed with the flag off: {span_line}" + ); + + let access_line = find_line(&captured, &["chimely::access", "/v1/subscribers/"]); + assert!( + access_line.contains("path=/v1/subscribers/usr_123"), + "access-log path changed with the flag off: {access_line}" + ); + + let stream_line = find_line(&captured, &["chimely::access", "/v1/inbox/stream"]); + assert!( + stream_line.contains("subscriber_id=usr_123"), + "subscriber_id changed with the flag off: {stream_line}" + ); + assert!( + stream_line.contains(&format!("environment={}", app.env.slug)), + "environment changed with the flag off: {stream_line}" + ); + assert!( + stream_line.contains("subscriber_hash=redacted"), + "{stream_line}" + ); + assert!( + !captured.contains(HASH_USR_123), + "identifier hashed despite the flag being off:\n{captured}" + ); +} diff --git a/server/tests/management.rs b/server/tests/management.rs index 57a483b..d0d0086 100644 --- a/server/tests/management.rs +++ b/server/tests/management.rs @@ -229,7 +229,9 @@ async fn replay_returns_the_snapshot_even_after_deliver_at_has_passed() { } /// S5 regression: extractor rejections must use the contract's error -/// envelope, never axum's plain-text 400/415/422. +/// envelope, never axum's plain-text 400/415/422. The message is static. +/// serde rejection prose can quote caller scalars, so no request byte may +/// reach the body (H4, #58). #[tokio::test] async fn malformed_bodies_get_the_error_envelope() { let app = support::spawn().await; @@ -245,6 +247,25 @@ async fn malformed_bodies_get_the_error_envelope() { assert_eq!(res.status(), 400); let body: serde_json::Value = res.json().await.expect("envelope body"); assert_eq!(body["error"]["code"], "invalid_request"); + assert_eq!(body["error"]["message"], "invalid request body"); + + // A type mismatch inside valid JSON must not echo the submitted scalar. + let res = app + .client + .post(format!("{}/v1/notifications", app.base)) + .bearer_auth(&app.env.api_key) + .json(&json!({ "subscriber_id": "usr_q", "category": 31337 })) + .send() + .await + .unwrap(); + assert_eq!(res.status(), 400); + let body: serde_json::Value = res.json().await.expect("envelope body"); + assert_eq!(body["error"]["code"], "invalid_request"); + assert_eq!(body["error"]["message"], "invalid request body"); + assert!( + !body["error"]["message"].as_str().unwrap().contains("31337"), + "rejection echoed caller input: {body}" + ); // Wrong content type is a 400 envelope too (the contract has no 415). let res = app @@ -259,6 +280,7 @@ async fn malformed_bodies_get_the_error_envelope() { assert_eq!(res.status(), 400); let body: serde_json::Value = res.json().await.expect("envelope body"); assert_eq!(body["error"]["code"], "invalid_request"); + assert_eq!(body["error"]["message"], "invalid request body"); // Malformed query parameters on the subscriber plane as well. let res = app @@ -271,6 +293,40 @@ async fn malformed_bodies_get_the_error_envelope() { assert_eq!(res.status(), 400); let body: serde_json::Value = res.json().await.expect("envelope body"); assert_eq!(body["error"]["code"], "invalid_request"); + assert_eq!(body["error"]["message"], "invalid query string"); + assert!( + !body["error"]["message"] + .as_str() + .unwrap() + .contains("banana"), + "rejection echoed caller input: {body}" + ); +} + +/// The optional-body extractor rejects with the same static message. +/// PUT /v1/subscribers/{id} takes `Option>`, so a malformed +/// body goes through `OptionalFromRequest`, whose rejection prose can also +/// quote caller scalars (H4, #58). +#[tokio::test] +async fn malformed_optional_bodies_get_the_static_message() { + let app = support::spawn().await; + let res = app + .client + .put(format!("{}/v1/subscribers/usr_opt", app.base)) + .bearer_auth(&app.env.api_key) + .json(&json!({ "created_at": 9001 })) + .send() + .await + .unwrap(); + assert_eq!(res.status(), 400); + let text = res.text().await.unwrap(); + assert!( + !text.contains("9001"), + "rejection echoed caller input: {text}" + ); + let body: serde_json::Value = serde_json::from_str(&text).expect("envelope body"); + assert_eq!(body["error"]["code"], "invalid_request"); + assert_eq!(body["error"]["message"], "invalid request body"); } #[tokio::test] diff --git a/server/tests/support/mod.rs b/server/tests/support/mod.rs index 3e7e36b..3a7df9e 100644 --- a/server/tests/support/mod.rs +++ b/server/tests/support/mod.rs @@ -149,6 +149,7 @@ async fn spawn_inner( subscriber_rate_burst: 0.0, shutdown_readiness_grace: Duration::from_millis(150), shutdown_drain_deadline: Duration::from_secs(5), + log_scrub_identifiers: false, }; configure(&mut cfg); let cfg = Arc::new(cfg);