diff --git a/CHANGELOG.md b/CHANGELOG.md index fcae3fdf763..3fddacdf899 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -12,6 +12,7 @@ **Bug Fixes**: +- 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)) - Integration endpoints now forward requests instead of silently dropping telemetry. ([#6611](https://github.com/getsentry/relay/pull/6611)) - 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) diff --git a/relay-replays/src/recording.rs b/relay-replays/src/recording.rs index 69b65e410e6..bd79105681f 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,58 @@ 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. +/// +/// 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(()); + } + decompress(body, limit, &mut io::sink()) +} + /// A utility that performs data scrubbing on compressed Replay recording payloads. /// /// ### Example @@ -349,11 +404,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 +428,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 +446,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, check_recording_size}; fn default_pii_config() -> PiiConfig { let mut scrubbing_config = DataScrubbingConfig::default(); @@ -682,6 +724,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_check_recording_size_within_limit() { + let body = b"[{\"type\":4}]"; + assert!(check_recording_size(&compressed_recording(body), body.len()).is_ok()); + } + + #[test] + fn test_check_recording_size_exceeds_limit() { + let body = b"[{\"type\":4}]"; + assert!(matches!( + check_recording_size(&compressed_recording(body), body.len() - 1), + Err(ParseRecordingError::TooLarge) + )); + } + + #[test] + fn test_check_recording_size_high_compression_ratio() { + let recording = compressed_recording(&vec![b' '; 1024 * 1024]); + assert!(recording.len() < 2048); + assert!(matches!( + check_recording_size(&recording, 1024), + Err(ParseRecordingError::TooLarge) + )); + } + + #[test] + fn test_check_recording_size_uncompressed() { + assert!(check_recording_size(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 130edb1043f..b38c99a95cc 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, ))), diff --git a/relay-server/src/processing/replays/process.rs b/relay-server/src/processing/replays/process.rs index 291cc13781b..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::RecordingScrubber; +use relay_replays::recording::{ParseRecordingError, RecordingScrubber, check_recording_size}; use relay_statsd::metric; use serde::Deserialize; @@ -136,11 +136,13 @@ 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> { let event_id = replay.headers.event_id(); + let limit = ctx.config.max_replay_uncompressed_size(); let pii_config = ctx .project_info .config @@ -148,18 +150,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) = check_recording_size(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 +190,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 +215,5 @@ pub fn scrub( } })?; - if ctx - .project_info - .has_feature(Feature::SessionReplayRecordingScrubbing) - { - scrub_recordings(replay, ctx) - } else { - Ok(()) - } + 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