From fd01d6ae15680c039cefc06398de1aa42b97098b Mon Sep 17 00:00:00 2001 From: "sentry-junior[bot]" <264270552+sentry-junior[bot]@users.noreply.github.com> Date: Wed, 28 Oct 2026 16:27:30 +0000 Subject: [PATCH 1/5] fix(replays): Bound recording uncompressed size without scrubbing Recording scrubbing already enforces max_replay_uncompressed_size, but only for projects that scrub recordings. Check the inflated size of every recording in processing Relays so the limit applies regardless. Refs REPLAY-994 Co-Authored-By: Dominik Buszowiecki <44422760+DominikB2014@users.noreply.github.com> --- CHANGELOG.md | 1 + relay-server/src/processing/replays/mod.rs | 16 +++- .../src/processing/replays/validate.rs | 79 +++++++++++++++++++ 3 files changed, 93 insertions(+), 3 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 19661be9c9c..3eb13753ff7 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -12,6 +12,7 @@ **Bug Fixes**: +- Reject replay recordings that decompress past `max_replay_uncompressed_size` in processing Relays, regardless of whether recording scrubbing is enabled. ([#PR](https://github.com/getsentry/relay/pull/PR)) - Preserve consistent quota and outcome quantities for empty attachments after expansion, preventing debug accounting panics. ([#6569](https://github.com/getsentry/relay/pull/6569)) - Use the dynamic sampling context to fill missing trace IDs on errors, preserving existing trace and span IDs. ([#6508](https://github.com/getsentry/relay/pull/6508)) - Fix Perfetto sample timestamps recorded in a clock other than the primary trace clock. ([#6522](https://github.com/getsentry/relay/pull/6522)) diff --git a/relay-server/src/processing/replays/mod.rs b/relay-server/src/processing/replays/mod.rs index 130edb1043f..b7d994c6c40 100644 --- a/relay-server/src/processing/replays/mod.rs +++ b/relay-server/src/processing/replays/mod.rs @@ -71,8 +71,7 @@ pub enum Error { #[cfg(feature = "processing")] #[error("failed to serialize replay")] FailedToSerializeReplay, - /// Replay recording too large for the consumer. - #[cfg(feature = "processing")] + /// Replay recording too large, either once decompressed or for the consumer message. #[error("replay recording too large")] TooLarge, /// The envelope did not contain an event ID. @@ -112,7 +111,6 @@ impl OutcomeError for Error { Self::Filtered(key) => Some(Outcome::Filtered(key.clone())), #[cfg(feature = "processing")] Self::FailedToSerializeReplay => Some(Outcome::Invalid(DiscardReason::Internal)), - #[cfg(feature = "processing")] Self::TooLarge => Some(Outcome::Invalid(DiscardReason::ItemTooLarge( crate::services::outcome::DiscardItemType::ReplayRecording, ))), @@ -191,6 +189,10 @@ impl processing::Processor for ReplaysProcessor { let mut replay = process::expand(replays)?; validate::validate(&replay).reject(&replay)?; + if ctx.is_processing() { + let max_size = ctx.config.max_replay_uncompressed_size(); + validate::recording_size(replay.payload.recording(), max_size).reject(&replay)?; + } process::normalize(&mut replay, &self.geoip_lookup, ctx)?; filter::filter(&replay, ctx).reject(&replay)?; @@ -315,6 +317,14 @@ impl ReplayPayload { } } + fn recording(&self) -> &Bytes { + match self { + ReplayPayload::WebReplay { recording, .. } => recording, + ReplayPayload::NativeReplay { recording, .. } => recording, + ReplayPayload::StandaloneRecording { recording } => recording, + } + } + fn recording_mut(&mut self) -> &mut Bytes { match self { ReplayPayload::WebReplay { recording, .. } => recording, diff --git a/relay-server/src/processing/replays/validate.rs b/relay-server/src/processing/replays/validate.rs index 49ba451304c..2313dcbb579 100644 --- a/relay-server/src/processing/replays/validate.rs +++ b/relay-server/src/processing/replays/validate.rs @@ -1,3 +1,6 @@ +use std::io::{self, Read}; + +use flate2::bufread::ZlibDecoder; use relay_event_normalization::replay; use crate::processing::replays::{Error, ExpandedReplay, Result}; @@ -14,3 +17,79 @@ pub fn validate(replay: &ExpandedReplay) -> Result<()> { let event = event.value().ok_or(Error::NoEventContent)?; replay::validate(event).map_err(Error::from) } + +/// Rejects a replay whose recording inflates to more than `max_size` bytes. +/// +/// Recording scrubbing applies the same bound, but only for projects which scrub recordings. +/// Without this check, any other project can store a small payload that inflates to an +/// arbitrary size on every read. Bodies which are not valid zlib streams are left to the +/// consumer, which already handles them. +pub fn recording_size(recording: &[u8], max_size: usize) -> Result<()> { + // A recording is a JSON header line followed by the body. + let body = recording + .splitn(2, |&b| b == b'\n') + .nth(1) + .unwrap_or_default(); + + // Reading one byte past the limit detects an oversized body without inflating it fully. + let mut decoder = ZlibDecoder::new(body).take(max_size as u64 + 1); + let size = io::copy(&mut decoder, &mut io::sink()).unwrap_or(0); + + if size > max_size as u64 { + relay_log::debug!("replay recording over maximum uncompressed size"); + return Err(Error::TooLarge); + } + + Ok(()) +} + +#[cfg(test)] +mod tests { + use std::io::Write; + + use flate2::Compression; + use flate2::write::ZlibEncoder; + + use super::*; + + fn recording(body: &[u8]) -> Vec { + let mut encoder = ZlibEncoder::new(Vec::new(), Compression::best()); + encoder.write_all(body).unwrap(); + let mut recording = b"{\"segment_id\":0}\n".to_vec(); + recording.extend(encoder.finish().unwrap()); + recording + } + + #[test] + fn test_recording_size_within_limit() { + let body = b"[{\"type\":4}]"; + assert!(recording_size(&recording(body), body.len()).is_ok()); + } + + #[test] + fn test_recording_size_exceeds_limit() { + let body = b"[{\"type\":4}]"; + assert!(matches!( + recording_size(&recording(body), body.len() - 1), + Err(Error::TooLarge) + )); + } + + #[test] + fn test_recording_size_high_compression_ratio() { + let body = vec![b' '; 1024 * 1024]; + let recording = recording(&body); + assert!(recording.len() < 2048); + assert!(matches!( + recording_size(&recording, 1024), + Err(Error::TooLarge) + )); + } + + #[test] + fn test_recording_size_ignores_uncompressed_and_invalid() { + assert!(recording_size(b"{\"segment_id\":0}\n[{\"type\":4}]", 1).is_ok()); + assert!(recording_size(b"{\"segment_id\":0}\nnot zlib", 1).is_ok()); + assert!(recording_size(b"no body", 1).is_ok()); + } +} From db985d1171d56074f74923cdede47a9d5acb11cc Mon Sep 17 00:00:00 2001 From: "sentry-junior[bot]" <264270552+sentry-junior[bot]@users.noreply.github.com> Date: Wed, 28 Oct 2026 16:27:57 +0000 Subject: [PATCH 2/5] add changelog PR number Co-Authored-By: Dominik Buszowiecki <44422760+DominikB2014@users.noreply.github.com> --- CHANGELOG.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 3eb13753ff7..5493625f797 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -12,7 +12,7 @@ **Bug Fixes**: -- Reject replay recordings that decompress past `max_replay_uncompressed_size` in processing Relays, regardless of whether recording scrubbing is enabled. ([#PR](https://github.com/getsentry/relay/pull/PR)) +- Reject replay recordings that decompress past `max_replay_uncompressed_size` in processing Relays, regardless of whether recording scrubbing is enabled. ([#6587](https://github.com/getsentry/relay/pull/6587)) - Preserve consistent quota and outcome quantities for empty attachments after expansion, preventing debug accounting panics. ([#6569](https://github.com/getsentry/relay/pull/6569)) - Use the dynamic sampling context to fill missing trace IDs on errors, preserving existing trace and span IDs. ([#6508](https://github.com/getsentry/relay/pull/6508)) - Fix Perfetto sample timestamps recorded in a clock other than the primary trace clock. ([#6522](https://github.com/getsentry/relay/pull/6522)) From 07ff5517261577dcd2f281a4a530e4a939e99d2a Mon Sep 17 00:00:00 2001 From: "sentry-junior[bot]" <264270552+sentry-junior[bot]@users.noreply.github.com> Date: Mon, 2 Nov 2026 20:21:03 +0000 Subject: [PATCH 3/5] move changelog entry to Unreleased --- CHANGELOG.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index aefe7188d2e..2b3a636162f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -12,6 +12,7 @@ **Bug Fixes**: +- Reject replay recordings that decompress past `max_replay_uncompressed_size` in processing Relays, regardless of whether recording scrubbing is enabled. ([#6587](https://github.com/getsentry/relay/pull/6587)) - Apply default replacement-only PII scrubbing to AI and MCP attributes on native spans. ([#6511](https://github.com/getsentry/relay/pull/6511), [#6612](https://github.com/getsentry/relay/pull/6612) **Internal**: @@ -35,7 +36,6 @@ **Bug Fixes**: -- Reject replay recordings that decompress past `max_replay_uncompressed_size` in processing Relays, regardless of whether recording scrubbing is enabled. ([#6587](https://github.com/getsentry/relay/pull/6587)) - Preserve consistent quota and outcome quantities for empty attachments after expansion, preventing debug accounting panics. ([#6569](https://github.com/getsentry/relay/pull/6569)) - Use the dynamic sampling context to fill missing trace IDs on errors, preserving existing trace and span IDs. ([#6508](https://github.com/getsentry/relay/pull/6508)) - Fix Perfetto sample timestamps recorded in a clock other than the primary trace clock. ([#6522](https://github.com/getsentry/relay/pull/6522)) From b1127fd8a6d1d0ec1652b0d8f693e5cf11bf26e5 Mon Sep 17 00:00:00 2001 From: "sentry-junior[bot]" <264270552+sentry-junior[bot]@users.noreply.github.com> Date: Tue, 3 Nov 2026 09:39:20 +0000 Subject: [PATCH 4/5] ref(replays): Enforce uncompressed recording size in the scrubbing pass Move the size check out of validation, which ran before filtering and quotas, into the recording scrubbing step. Scrubbing now reports an explicit TooLarge error from its own decompression, and processing Relays without recording scrubbing run the same decompression helper to check the size, so each recording is decompressed at most once. --- CHANGELOG.md | 2 +- relay-replays/src/recording.rs | 129 +++++++++++++++--- relay-server/src/processing/replays/mod.rs | 12 -- .../src/processing/replays/process.rs | 55 +++++--- .../src/processing/replays/validate.rs | 79 ----------- 5 files changed, 142 insertions(+), 135 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 2b3a636162f..6d7020273ae 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -12,7 +12,7 @@ **Bug Fixes**: -- Reject replay recordings that decompress past `max_replay_uncompressed_size` in processing Relays, regardless of whether recording scrubbing is enabled. ([#6587](https://github.com/getsentry/relay/pull/6587)) +- Reject replay recordings that decompress past `max_replay_uncompressed_size` with an `ItemTooLarge` outcome, also in processing Relays for projects without recording scrubbing. ([#6587](https://github.com/getsentry/relay/pull/6587)) - Apply default replacement-only PII scrubbing to AI and MCP attributes on native spans. ([#6511](https://github.com/getsentry/relay/pull/6511), [#6612](https://github.com/getsentry/relay/pull/6612) **Internal**: diff --git a/relay-replays/src/recording.rs b/relay-replays/src/recording.rs index 69b65e410e6..f4c9a331efc 100644 --- a/relay-replays/src/recording.rs +++ b/relay-replays/src/recording.rs @@ -13,7 +13,7 @@ use std::cell::RefCell; use std::fmt; -use std::io::Read; +use std::io::{self, Read, Write}; use std::rc::Rc; use std::{borrow::Cow, sync::LazyLock}; @@ -58,6 +58,8 @@ pub enum ParseRecordingError { Parse(serde_json::Error), /// Invalid or broken compression. Compression(std::io::Error), + /// The payload exceeds the size limit after decompression. + TooLarge, /// Validation of the payload failed. /// /// The body is empty, is missing the headers, or the body. @@ -69,6 +71,7 @@ impl fmt::Display for ParseRecordingError { match self { ParseRecordingError::Parse(serde_error) => write!(f, "{serde_error}"), ParseRecordingError::Compression(error) => write!(f, "{error}"), + ParseRecordingError::TooLarge => write!(f, "payload too large after decompression"), ParseRecordingError::Message(message) => write!(f, "{message}"), } } @@ -79,7 +82,7 @@ impl std::error::Error for ParseRecordingError { match self { ParseRecordingError::Parse(e) => Some(e), ParseRecordingError::Compression(e) => Some(e), - ParseRecordingError::Message(_) => None, + ParseRecordingError::TooLarge | ParseRecordingError::Message(_) => None, } } } @@ -268,6 +271,52 @@ where D::custom(s.to_string()) } +/// Splits a recording payload into its header line and body. +fn split_recording(bytes: &[u8]) -> Result<(&[u8], &[u8]), ParseRecordingError> { + if bytes.is_empty() { + return Err(ParseRecordingError::Message("no data found")); + } + + let mut split = bytes.splitn(2, |b| b == &b'\n'); + let header = split + .next() + .ok_or(ParseRecordingError::Message("no headers found"))?; + + match split.next() { + Some(b"") | None => Err(ParseRecordingError::Message("no body found")), + Some(body) => Ok((header, body)), + } +} + +/// Inflates a zlib compressed recording body into `output`. +/// +/// Reads at most one byte past `limit`, so oversized bodies are detected without inflating them +/// fully. +fn decompress( + body: &[u8], + limit: usize, + output: &mut impl Write, +) -> Result<(), ParseRecordingError> { + let mut decoder = ZlibDecoder::new(body).take((limit as u64).saturating_add(1)); + let size = io::copy(&mut decoder, output).map_err(ParseRecordingError::Compression)?; + if size > limit as u64 { + return Err(ParseRecordingError::TooLarge); + } + Ok(()) +} + +/// Checks that a recording payload does not exceed `limit` bytes after decompression. +/// +/// This applies the same decompression and limit as [`RecordingScrubber::process_recording`], +/// for recordings which are not scrubbed. Uncompressed bodies are not checked. +pub fn validate_recording(bytes: &[u8], limit: usize) -> Result<(), ParseRecordingError> { + let (_, body) = split_recording(bytes)?; + if body.first() == Some(&b'[') { + return Ok(()); + } + decompress(body, limit, &mut io::sink()) +} + /// A utility that performs data scrubbing on compressed Replay recording payloads. /// /// ### Example @@ -349,11 +398,7 @@ impl<'a> RecordingScrubber<'a> { self.scrub_replay(body, encoder) } else { let mut decompressed = Vec::with_capacity(8 * 1024); - let mut decoder = ZlibDecoder::new(body).take(self.limit as u64); - decoder - .read_to_end(&mut decompressed) - .map_err(ParseRecordingError::Compression)?; - + decompress(body, self.limit, &mut decompressed)?; self.scrub_replay(&decompressed, encoder) } } @@ -377,20 +422,7 @@ impl<'a> RecordingScrubber<'a> { /// - The payload size exceeds the configured `limit` of the scrubber after decompression. /// - On errors during decompression or JSON parsing. pub fn process_recording(&mut self, bytes: &[u8]) -> Result, ParseRecordingError> { - // Check for null byte condition. - if bytes.is_empty() { - return Err(ParseRecordingError::Message("no data found")); - } - - let mut split = bytes.splitn(2, |b| b == &b'\n'); - let header = split - .next() - .ok_or(ParseRecordingError::Message("no headers found"))?; - - let body = match split.next() { - Some(b"") | None => return Err(ParseRecordingError::Message("no body found")), - Some(body) => body, - }; + let (header, body) = split_recording(bytes)?; let mut output = header.to_owned(); output.push(b'\n'); @@ -408,11 +440,15 @@ impl<'a> RecordingScrubber<'a> { mod tests { // End to end test coverage. + use std::io::Write; + + use flate2::Compression; + use flate2::write::ZlibEncoder; use relay_pii::{DataScrubbingConfig, PiiConfig}; use crate::recording::scrub_at_path; - use super::RecordingScrubber; + use super::{ParseRecordingError, RecordingScrubber, validate_recording}; fn default_pii_config() -> PiiConfig { let mut scrubbing_config = DataScrubbingConfig::default(); @@ -682,6 +718,55 @@ mod tests { ); } + fn compressed_recording(body: &[u8]) -> Vec { + let mut encoder = ZlibEncoder::new(Vec::new(), Compression::best()); + encoder.write_all(body).unwrap(); + let mut recording = b"{\"segment_id\":0}\n".to_vec(); + recording.extend(encoder.finish().unwrap()); + recording + } + + #[test] + fn test_validate_recording_within_limit() { + let body = b"[{\"type\":4}]"; + assert!(validate_recording(&compressed_recording(body), body.len()).is_ok()); + } + + #[test] + fn test_validate_recording_exceeds_limit() { + let body = b"[{\"type\":4}]"; + assert!(matches!( + validate_recording(&compressed_recording(body), body.len() - 1), + Err(ParseRecordingError::TooLarge) + )); + } + + #[test] + fn test_validate_recording_high_compression_ratio() { + let recording = compressed_recording(&vec![b' '; 1024 * 1024]); + assert!(recording.len() < 2048); + assert!(matches!( + validate_recording(&recording, 1024), + Err(ParseRecordingError::TooLarge) + )); + } + + #[test] + fn test_validate_recording_uncompressed() { + assert!(validate_recording(b"{\"segment_id\":0}\n[{\"type\":4}]", 1).is_ok()); + } + + #[test] + fn test_process_recording_exceeds_limit() { + let body = b"[{\"type\":4}]"; + let config = default_pii_config(); + let mut scrubber = RecordingScrubber::new(body.len() - 1, Some(&config), None); + assert!(matches!( + scrubber.process_recording(&compressed_recording(body)), + Err(ParseRecordingError::TooLarge) + )); + } + #[test] fn test_scrub_at_path() { for (should_scrub, path) in [ diff --git a/relay-server/src/processing/replays/mod.rs b/relay-server/src/processing/replays/mod.rs index b7d994c6c40..b38c99a95cc 100644 --- a/relay-server/src/processing/replays/mod.rs +++ b/relay-server/src/processing/replays/mod.rs @@ -189,10 +189,6 @@ impl processing::Processor for ReplaysProcessor { let mut replay = process::expand(replays)?; validate::validate(&replay).reject(&replay)?; - if ctx.is_processing() { - let max_size = ctx.config.max_replay_uncompressed_size(); - validate::recording_size(replay.payload.recording(), max_size).reject(&replay)?; - } process::normalize(&mut replay, &self.geoip_lookup, ctx)?; filter::filter(&replay, ctx).reject(&replay)?; @@ -317,14 +313,6 @@ impl ReplayPayload { } } - fn recording(&self) -> &Bytes { - match self { - ReplayPayload::WebReplay { recording, .. } => recording, - ReplayPayload::NativeReplay { recording, .. } => recording, - ReplayPayload::StandaloneRecording { recording } => recording, - } - } - fn recording_mut(&mut self) -> &mut Bytes { match self { ReplayPayload::WebReplay { recording, .. } => recording, diff --git a/relay-server/src/processing/replays/process.rs b/relay-server/src/processing/replays/process.rs index 291cc13781b..09718c5bd88 100644 --- a/relay-server/src/processing/replays/process.rs +++ b/relay-server/src/processing/replays/process.rs @@ -4,7 +4,7 @@ use relay_event_schema::processor::{self, ProcessingState}; use relay_event_schema::protocol::Replay; use relay_pii::PiiProcessor; use relay_protocol::Annotated; -use relay_replays::recording::RecordingScrubber; +use relay_replays::recording::{ParseRecordingError, RecordingScrubber, validate_recording}; use relay_statsd::metric; use serde::Deserialize; @@ -141,6 +141,7 @@ fn scrub_recordings( ctx: Context<'_>, ) -> Result<(), Rejected> { let event_id = replay.headers.event_id(); + let limit = ctx.config.max_replay_uncompressed_size(); let pii_config = ctx .project_info .config @@ -148,18 +149,34 @@ fn scrub_recordings( .pii_config() .as_ref(); - let mut scrubber = RecordingScrubber::new( - ctx.config.max_replay_uncompressed_size(), - ctx.project_info.config.pii_config.as_ref(), - pii_config, - ); - - if scrubber.is_empty() { - return Ok(()); - } + let mut scrubber = ctx + .project_info + .has_feature(Feature::SessionReplayRecordingScrubbing) + .then(|| { + RecordingScrubber::new( + limit, + ctx.project_info.config.pii_config.as_ref(), + pii_config, + ) + }) + .filter(|scrubber| !scrubber.is_empty()); replay.try_modify(|replay, _| { let payload = replay.payload.recording_mut(); + + let Some(scrubber) = scrubber.as_mut() else { + // Scrubbing enforces the limit while decompressing. Without it, processing Relays + // run the same decompression only to check the size. Other malformed recordings + // are left to the consumer. + if ctx.is_processing() + && let Err(ParseRecordingError::TooLarge) = validate_recording(payload, limit) + { + relay_log::debug!("replay recording over maximum uncompressed size"); + return Err(Error::TooLarge); + } + return Ok(()); + }; + *payload = metric!(timer(RelayTimers::ReplayRecordingProcessing), { scrubber.process_recording(payload) }) @@ -172,15 +189,18 @@ fn scrub_recordings( organization_id = ctx.project_info.organization_id.map(|o| o.value()), "invalid replay recording" ); - Error::InvalidReplayRecordingEvent + match error { + ParseRecordingError::TooLarge => Error::TooLarge, + _ => Error::InvalidReplayRecordingEvent, + } })?; - Ok::<(), Error>(()) + Ok(()) }) } /// Applies PII scrubbing to a replay event and recording. /// -/// Will reject if the scrubbing fails. +/// Will reject if the scrubbing fails, or if the recording exceeds the maximum uncompressed size. pub fn scrub( replay: &mut Managed, ctx: Context<'_>, @@ -194,12 +214,5 @@ pub fn scrub( } })?; - if ctx - .project_info - .has_feature(Feature::SessionReplayRecordingScrubbing) - { - scrub_recordings(replay, ctx) - } else { - Ok(()) - } + scrub_recordings(replay, ctx) } diff --git a/relay-server/src/processing/replays/validate.rs b/relay-server/src/processing/replays/validate.rs index 2313dcbb579..49ba451304c 100644 --- a/relay-server/src/processing/replays/validate.rs +++ b/relay-server/src/processing/replays/validate.rs @@ -1,6 +1,3 @@ -use std::io::{self, Read}; - -use flate2::bufread::ZlibDecoder; use relay_event_normalization::replay; use crate::processing::replays::{Error, ExpandedReplay, Result}; @@ -17,79 +14,3 @@ pub fn validate(replay: &ExpandedReplay) -> Result<()> { let event = event.value().ok_or(Error::NoEventContent)?; replay::validate(event).map_err(Error::from) } - -/// Rejects a replay whose recording inflates to more than `max_size` bytes. -/// -/// Recording scrubbing applies the same bound, but only for projects which scrub recordings. -/// Without this check, any other project can store a small payload that inflates to an -/// arbitrary size on every read. Bodies which are not valid zlib streams are left to the -/// consumer, which already handles them. -pub fn recording_size(recording: &[u8], max_size: usize) -> Result<()> { - // A recording is a JSON header line followed by the body. - let body = recording - .splitn(2, |&b| b == b'\n') - .nth(1) - .unwrap_or_default(); - - // Reading one byte past the limit detects an oversized body without inflating it fully. - let mut decoder = ZlibDecoder::new(body).take(max_size as u64 + 1); - let size = io::copy(&mut decoder, &mut io::sink()).unwrap_or(0); - - if size > max_size as u64 { - relay_log::debug!("replay recording over maximum uncompressed size"); - return Err(Error::TooLarge); - } - - Ok(()) -} - -#[cfg(test)] -mod tests { - use std::io::Write; - - use flate2::Compression; - use flate2::write::ZlibEncoder; - - use super::*; - - fn recording(body: &[u8]) -> Vec { - let mut encoder = ZlibEncoder::new(Vec::new(), Compression::best()); - encoder.write_all(body).unwrap(); - let mut recording = b"{\"segment_id\":0}\n".to_vec(); - recording.extend(encoder.finish().unwrap()); - recording - } - - #[test] - fn test_recording_size_within_limit() { - let body = b"[{\"type\":4}]"; - assert!(recording_size(&recording(body), body.len()).is_ok()); - } - - #[test] - fn test_recording_size_exceeds_limit() { - let body = b"[{\"type\":4}]"; - assert!(matches!( - recording_size(&recording(body), body.len() - 1), - Err(Error::TooLarge) - )); - } - - #[test] - fn test_recording_size_high_compression_ratio() { - let body = vec![b' '; 1024 * 1024]; - let recording = recording(&body); - assert!(recording.len() < 2048); - assert!(matches!( - recording_size(&recording, 1024), - Err(Error::TooLarge) - )); - } - - #[test] - fn test_recording_size_ignores_uncompressed_and_invalid() { - assert!(recording_size(b"{\"segment_id\":0}\n[{\"type\":4}]", 1).is_ok()); - assert!(recording_size(b"{\"segment_id\":0}\nnot zlib", 1).is_ok()); - assert!(recording_size(b"no body", 1).is_ok()); - } -} From af19471732bd75f7b5365b2dc8e85b57d974988b Mon Sep 17 00:00:00 2001 From: "sentry-junior[bot]" <264270552+sentry-junior[bot]@users.noreply.github.com> Date: Tue, 3 Nov 2026 10:21:21 +0000 Subject: [PATCH 5/5] ref(replays): Rename recording size check and add integration tests Rename validate_recording to check_recording_size and document that callers may pass malformed recordings through, and rename scrub_recordings to process_recordings since it also enforces the size limit without scrubbing. Add integration tests for processing Relays without recording scrubbing and without a PII config, covering recordings at and over max_replay_uncompressed_size. --- relay-replays/src/recording.rs | 26 ++++--- .../src/processing/replays/process.rs | 9 +-- tests/integration/test_replay_recordings.py | 69 +++++++++++++++++++ 3 files changed, 90 insertions(+), 14 deletions(-) diff --git a/relay-replays/src/recording.rs b/relay-replays/src/recording.rs index f4c9a331efc..bd79105681f 100644 --- a/relay-replays/src/recording.rs +++ b/relay-replays/src/recording.rs @@ -309,7 +309,13 @@ fn decompress( /// /// This applies the same decompression and limit as [`RecordingScrubber::process_recording`], /// for recordings which are not scrubbed. Uncompressed bodies are not checked. -pub fn validate_recording(bytes: &[u8], limit: usize) -> Result<(), ParseRecordingError> { +/// +/// This is only a size check, not a validation of the recording. It returns +/// [`ParseRecordingError::TooLarge`] for oversized payloads, but it also returns the other +/// errors it hits on the way, such as a missing body or a broken zlib stream. Callers which +/// only enforce the size limit can pass such malformed recordings through and leave them to +/// downstream consumers, as Relay does for recordings which are not scrubbed. +pub fn check_recording_size(bytes: &[u8], limit: usize) -> Result<(), ParseRecordingError> { let (_, body) = split_recording(bytes)?; if body.first() == Some(&b'[') { return Ok(()); @@ -448,7 +454,7 @@ mod tests { use crate::recording::scrub_at_path; - use super::{ParseRecordingError, RecordingScrubber, validate_recording}; + use super::{ParseRecordingError, RecordingScrubber, check_recording_size}; fn default_pii_config() -> PiiConfig { let mut scrubbing_config = DataScrubbingConfig::default(); @@ -727,33 +733,33 @@ mod tests { } #[test] - fn test_validate_recording_within_limit() { + fn test_check_recording_size_within_limit() { let body = b"[{\"type\":4}]"; - assert!(validate_recording(&compressed_recording(body), body.len()).is_ok()); + assert!(check_recording_size(&compressed_recording(body), body.len()).is_ok()); } #[test] - fn test_validate_recording_exceeds_limit() { + fn test_check_recording_size_exceeds_limit() { let body = b"[{\"type\":4}]"; assert!(matches!( - validate_recording(&compressed_recording(body), body.len() - 1), + check_recording_size(&compressed_recording(body), body.len() - 1), Err(ParseRecordingError::TooLarge) )); } #[test] - fn test_validate_recording_high_compression_ratio() { + fn test_check_recording_size_high_compression_ratio() { let recording = compressed_recording(&vec![b' '; 1024 * 1024]); assert!(recording.len() < 2048); assert!(matches!( - validate_recording(&recording, 1024), + check_recording_size(&recording, 1024), Err(ParseRecordingError::TooLarge) )); } #[test] - fn test_validate_recording_uncompressed() { - assert!(validate_recording(b"{\"segment_id\":0}\n[{\"type\":4}]", 1).is_ok()); + fn test_check_recording_size_uncompressed() { + assert!(check_recording_size(b"{\"segment_id\":0}\n[{\"type\":4}]", 1).is_ok()); } #[test] diff --git a/relay-server/src/processing/replays/process.rs b/relay-server/src/processing/replays/process.rs index 09718c5bd88..6d284cd7c28 100644 --- a/relay-server/src/processing/replays/process.rs +++ b/relay-server/src/processing/replays/process.rs @@ -4,7 +4,7 @@ use relay_event_schema::processor::{self, ProcessingState}; use relay_event_schema::protocol::Replay; use relay_pii::PiiProcessor; use relay_protocol::Annotated; -use relay_replays::recording::{ParseRecordingError, RecordingScrubber, validate_recording}; +use relay_replays::recording::{ParseRecordingError, RecordingScrubber, check_recording_size}; use relay_statsd::metric; use serde::Deserialize; @@ -136,7 +136,8 @@ fn scrub_event(event: &mut Annotated, ctx: Context<'_>) -> Result<(), Er Ok(()) } -fn scrub_recordings( +/// Scrubs a replay recording if configured and enforces its maximum uncompressed size. +fn process_recordings( replay: &mut Managed, ctx: Context<'_>, ) -> Result<(), Rejected> { @@ -169,7 +170,7 @@ fn scrub_recordings( // run the same decompression only to check the size. Other malformed recordings // are left to the consumer. if ctx.is_processing() - && let Err(ParseRecordingError::TooLarge) = validate_recording(payload, limit) + && let Err(ParseRecordingError::TooLarge) = check_recording_size(payload, limit) { relay_log::debug!("replay recording over maximum uncompressed size"); return Err(Error::TooLarge); @@ -214,5 +215,5 @@ pub fn scrub( } })?; - scrub_recordings(replay, ctx) + process_recordings(replay, ctx) } diff --git a/tests/integration/test_replay_recordings.py b/tests/integration/test_replay_recordings.py index 77e2c7c0eb9..c753313bd2b 100644 --- a/tests/integration/test_replay_recordings.py +++ b/tests/integration/test_replay_recordings.py @@ -1,7 +1,12 @@ +import json import zlib +import pytest +from sentry_relay.consts import DataCategory from sentry_sdk.envelope import Envelope, Item, PayloadRef +from .asserts import time_within_delta +from .consts import Outcome from .test_replay_events import generate_replay_sdk_event @@ -87,6 +92,70 @@ def test_nonchunked_replay_recordings_processing( outcomes_consumer.assert_empty() +@pytest.mark.parametrize( + "project_config", + [ + pytest.param( + {"features": ["organizations:session-replay"]}, + id="scrubbing_disabled", + ), + pytest.param( + { + "features": [ + "organizations:session-replay", + "organizations:session-replay-recording-scrubbing", + ], + "piiConfig": None, + }, + id="no_pii_config", + ), + ], +) +@pytest.mark.parametrize("too_large", [False, True], ids=["at_limit", "over_limit"]) +def test_replay_recording_max_uncompressed_size( + mini_sentry, + relay_with_processing, + replay_recordings_consumer, + outcomes_consumer, + project_config, + too_large, +): + project_id = 42 + replay_id = "515539018c9b4260a6f999572f1661ee" + + body = json.dumps([{"type": 4, "data": {"href": "a" * 1000}}]).encode() + limit = len(body) - 1 if too_large else len(body) + + mini_sentry.add_basic_project_config(project_id, extra={"config": project_config}) + relay = relay_with_processing({"limits": {"max_replay_uncompressed_size": limit}}) + replay_recordings_consumer = replay_recordings_consumer() + outcomes_consumer = outcomes_consumer() + + envelope = Envelope(headers=[["event_id", replay_id]]) + payload = recording_payload(body) + envelope.add_item(Item(payload=PayloadRef(bytes=payload), type="replay_recording")) + + relay.send_envelope(project_id, envelope) + + if too_large: + outcome = outcomes_consumer.get_outcome() + assert outcome == { + "timestamp": time_within_delta(), + "project_id": project_id, + "key_id": 123, + "outcome": Outcome.INVALID, + "reason": "too_large:replay_recording", + "category": DataCategory.REPLAY, + "quantity": 1, + } + replay_recordings_consumer.assert_empty() + else: + replay_recording = replay_recordings_consumer.get_not_chunked_replay(timeout=10) + assert replay_recording["replay_id"] == replay_id + assert replay_recording["payload"] == payload + outcomes_consumer.assert_empty() + + def recording_payload(bits: bytes): compressed_payload = zlib.compress(bits) return b'{"segment_id": 0}\n' + compressed_payload