relay_event_normalization/
replay.rs

1//! Validation and normalization of [`Replay`] events.
2
3use std::net::IpAddr as StdIpAddr;
4
5use relay_event_schema::processor::{self, ProcessingState, Processor};
6use relay_event_schema::protocol::{Contexts, IpAddr, Replay};
7use relay_protocol::Annotated;
8
9use crate::event::normalize_user_geoinfo;
10use crate::normalize::user_agent;
11use crate::user_agent::RawUserAgentInfo;
12use crate::{GeoIpLookup, trimming};
13
14/// Replay validation error.
15///
16/// This error is returned from [`validate`].
17#[derive(Debug, thiserror::Error)]
18pub enum ReplayError {
19    /// The replay event is missing a `replay_id`.
20    #[error("missing replay_id")]
21    MissingReplayId,
22    /// The replay event is missing a `segment_id`.
23    #[error("missing segment_id")]
24    MissingSegmentId,
25    /// The `segment_id` is to large to fit in a a u16.
26    #[error("segment_id too large")]
27    SegmentIdTooLarge,
28    /// One or more of the `error_ids` have an error.
29    #[error("invalid error_id specified")]
30    InvalidErrorId,
31    /// One or more of the `trace_ids` have an error.
32    #[error("invalid trace_id specified")]
33    InvalidTraceId,
34}
35
36/// Checks if the Replay event is structurally valid.
37///
38/// Returns `Ok(())`, if the Replay is valid and can be normalized. Otherwise, returns
39/// `Err(ReplayError::InvalidPayload)` describing the missing or invalid data.
40pub fn validate(replay: &Replay) -> Result<(), ReplayError> {
41    replay
42        .replay_id
43        .value()
44        .ok_or(ReplayError::MissingReplayId)?;
45
46    let segment_id = *replay
47        .segment_id
48        .value()
49        .ok_or(ReplayError::MissingSegmentId)?;
50
51    if segment_id > u16::MAX as u64 {
52        return Err(ReplayError::SegmentIdTooLarge);
53    }
54
55    if replay
56        .error_ids
57        .value()
58        .into_iter()
59        .flat_map(|v| v.iter())
60        .any(|v| v.meta().has_errors())
61    {
62        return Err(ReplayError::InvalidErrorId);
63    }
64
65    if replay
66        .trace_ids
67        .value()
68        .into_iter()
69        .flat_map(|v| v.iter())
70        .any(|v| v.meta().has_errors())
71    {
72        return Err(ReplayError::InvalidTraceId);
73    }
74
75    Ok(())
76}
77
78/// Adds default fields and normalizes all values in to their standard representation.
79pub fn normalize(
80    replay: &mut Annotated<Replay>,
81    client_ip: Option<StdIpAddr>,
82    user_agent: &RawUserAgentInfo<&str>,
83    geoip_lookup: &GeoIpLookup,
84) {
85    let _ = processor::apply(replay, |replay_value, meta| {
86        normalize_platform(replay_value);
87        normalize_ip_address(replay_value, client_ip);
88        normalize_user_geoinfo(
89            geoip_lookup,
90            &mut replay_value.user,
91            client_ip.map(|ip| IpAddr(ip.to_string())).as_ref(),
92        );
93        normalize_user_agent(replay_value, user_agent);
94        normalize_type(replay_value);
95        normalize_array_fields(replay_value);
96        let _ = trimming::TrimmingProcessor::new().process_replay(
97            replay_value,
98            meta,
99            ProcessingState::root(),
100        );
101        Ok(())
102    });
103}
104
105fn normalize_array_fields(replay: &mut Replay) {
106    // TODO: This should be replaced by the TrimmingProcessor.
107    // https://github.com/getsentry/relay/pull/1910#pullrequestreview-1337188206
108    if let Some(items) = replay.error_ids.value_mut() {
109        items.truncate(100);
110    }
111
112    if let Some(items) = replay.trace_ids.value_mut() {
113        items.truncate(100);
114    }
115
116    if let Some(items) = replay.urls.value_mut() {
117        items.truncate(100);
118    }
119    if let Some(items) = replay.segment_names.value_mut() {
120        items.truncate(100);
121    }
122}
123
124fn normalize_ip_address(replay: &mut Replay, ip_address: Option<StdIpAddr>) {
125    crate::event::normalize_ip_addresses(
126        &mut replay.request,
127        &mut replay.user,
128        replay.platform.as_str(),
129        ip_address.map(|ip| IpAddr(ip.to_string())).as_ref(),
130        replay.sdk.value(),
131    );
132}
133
134fn normalize_user_agent(replay: &mut Replay, default_user_agent: &RawUserAgentInfo<&str>) {
135    let headers = match replay
136        .request
137        .value()
138        .and_then(|request| request.headers.value())
139    {
140        Some(headers) => headers,
141        None => return,
142    };
143
144    let user_agent_info = RawUserAgentInfo::from_headers(headers);
145    let user_agent_info = if user_agent_info.is_empty() {
146        default_user_agent
147    } else {
148        &user_agent_info
149    };
150
151    let contexts = replay.contexts.get_or_insert_with(Contexts::new);
152    user_agent::normalize_user_agent_info_generic(contexts, &replay.platform, user_agent_info);
153}
154
155fn normalize_platform(replay: &mut Replay) {
156    // Null platforms are permitted but must be defaulted before continuing.
157    let platform = replay.platform.get_or_insert_with(|| "other".to_owned());
158
159    // Normalize bad platforms to "other" type.
160    if !crate::is_valid_platform(platform) {
161        replay.platform = Annotated::from("other".to_owned());
162    }
163}
164
165fn normalize_type(replay: &mut Replay) {
166    replay.ty = Annotated::from("replay_event".to_owned());
167}
168
169#[cfg(test)]
170mod tests {
171    use std::net::{IpAddr, Ipv4Addr};
172
173    use chrono::{TimeZone, Utc};
174    use insta::assert_json_snapshot;
175    use relay_protocol::{SerializableAnnotated, assert_annotated_snapshot, get_value};
176    use uuid::Uuid;
177
178    use relay_event_schema::protocol::{
179        BrowserContext, Context, DeviceContext, EventId, OsContext, TagEntry, Tags,
180    };
181
182    use super::*;
183
184    #[test]
185    fn test_event_roundtrip() {
186        // NOTE: Interfaces will be tested separately.
187        let json = r#"{
188  "event_id": "52df9022835246eeb317dbd739ccd059",
189  "replay_id": "52df9022835246eeb317dbd739ccd059",
190  "segment_id": 0,
191  "replay_type": "session",
192  "error_sample_rate": 0.5,
193  "session_sample_rate": 0.5,
194  "timestamp": 946684800.0,
195  "replay_start_timestamp": 946684800.0,
196  "urls": ["localhost:9000"],
197  "error_ids": ["52df9022835246eeb317dbd739ccd059"],
198  "trace_ids": ["52df9022835246eeb317dbd739ccd059"],
199  "platform": "myplatform",
200  "release": "myrelease",
201  "dist": "mydist",
202  "environment": "myenv",
203  "tags": [
204    [
205      "tag",
206      "value"
207    ]
208  ]
209}"#;
210
211        let replay = Annotated::new(Replay {
212            event_id: Annotated::new(EventId("52df9022835246eeb317dbd739ccd059".parse().unwrap())),
213            replay_id: Annotated::new(EventId("52df9022835246eeb317dbd739ccd059".parse().unwrap())),
214            replay_type: Annotated::new("session".to_owned()),
215            segment_id: Annotated::new(0),
216            timestamp: Annotated::new(Utc.with_ymd_and_hms(2000, 1, 1, 0, 0, 0).unwrap().into()),
217            replay_start_timestamp: Annotated::new(
218                Utc.with_ymd_and_hms(2000, 1, 1, 0, 0, 0).unwrap().into(),
219            ),
220            urls: Annotated::new(vec![Annotated::new("localhost:9000".to_owned())]),
221            error_ids: Annotated::new(vec![Annotated::new(
222                Uuid::parse_str("52df9022835246eeb317dbd739ccd059").unwrap(),
223            )]),
224            trace_ids: Annotated::new(vec![Annotated::new(
225                Uuid::parse_str("52df9022835246eeb317dbd739ccd059").unwrap(),
226            )]),
227            platform: Annotated::new("myplatform".to_owned()),
228            release: Annotated::new("myrelease".to_owned().into()),
229            dist: Annotated::new("mydist".to_owned()),
230            environment: Annotated::new("myenv".to_owned()),
231            tags: {
232                let items = vec![Annotated::new(TagEntry(
233                    Annotated::new("tag".to_owned()),
234                    Annotated::new("value".to_owned()),
235                ))];
236                Annotated::new(Tags(items.into()))
237            },
238            ..Default::default()
239        });
240
241        assert_eq!(replay, Annotated::from_json(json).unwrap());
242    }
243
244    #[test]
245    fn test_lenient_release() {
246        let input = r#"{"release":42}"#;
247        let output = r#"{"release":"42"}"#;
248        let event = Annotated::new(Replay {
249            release: Annotated::new("42".to_owned().into()),
250            ..Default::default()
251        });
252
253        assert_eq!(event, Annotated::from_json(input).unwrap());
254        assert_eq!(output, event.to_json().unwrap());
255    }
256
257    #[test]
258    fn test_set_user_agent_meta() {
259        // Parse user input.
260        let payload = include_str!("../../tests/fixtures/replay.json");
261
262        let mut replay: Annotated<Replay> = Annotated::from_json(payload).unwrap();
263        normalize(
264            &mut replay,
265            None,
266            &RawUserAgentInfo::default(),
267            &GeoIpLookup::empty(),
268        );
269
270        let contexts = get_value!(replay.contexts!);
271        assert_eq!(
272            contexts.get::<BrowserContext>(),
273            Some(&BrowserContext {
274                name: Annotated::new("Safari".to_owned()),
275                version: Annotated::new("15.5".to_owned()),
276                ..Default::default()
277            })
278        );
279        assert_eq!(
280            contexts.get_key("client_os"),
281            Some(&Context::Os(Box::new(OsContext {
282                name: Annotated::new("Mac OS X".to_owned()),
283                version: Annotated::new(">=10.15.7".to_owned()),
284                ..Default::default()
285            })))
286        );
287        assert_eq!(
288            contexts.get::<DeviceContext>(),
289            Some(&DeviceContext {
290                family: Annotated::new("Mac".to_owned()),
291                brand: Annotated::new("Apple".to_owned()),
292                model: Annotated::new("Mac".to_owned()),
293                ..Default::default()
294            })
295        );
296    }
297
298    #[test]
299    fn test_missing_user() {
300        let payload = include_str!("../../tests/fixtures/replay_missing_user.json");
301
302        let mut replay: Annotated<Replay> = Annotated::from_json(payload).unwrap();
303
304        // No user object and no ip-address was provided.
305        normalize(
306            &mut replay,
307            None,
308            &RawUserAgentInfo::default(),
309            &GeoIpLookup::empty(),
310        );
311        assert_eq!(get_value!(replay.user.geo), None);
312
313        // No user object but an ip-address was provided.
314        let ip_address = IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1));
315        normalize(
316            &mut replay,
317            Some(ip_address),
318            &RawUserAgentInfo::default(),
319            &GeoIpLookup::empty(),
320        );
321
322        let ipaddr = get_value!(replay.user!).ip_address.as_str();
323        assert_eq!(Some("127.0.0.1"), ipaddr);
324    }
325
326    #[test]
327    fn test_set_ip_address_missing_user_ip_address_and_geo() {
328        let lookup = GeoIpLookup::open("tests/fixtures/GeoIP2-Enterprise-Test.mmdb").unwrap();
329        let ip_address = IpAddr::V4(Ipv4Addr::new(2, 125, 160, 216));
330
331        // IP-Address set.
332        let payload = include_str!("../../tests/fixtures/replay_missing_user_ip_address.json");
333
334        let mut replay: Annotated<Replay> = Annotated::from_json(payload).unwrap();
335        normalize(
336            &mut replay,
337            Some(ip_address),
338            &RawUserAgentInfo::default(),
339            &lookup,
340        );
341
342        let user = &replay.value().unwrap().user;
343        assert_json_snapshot!(SerializableAnnotated(user), @r###"
344        {
345          "id": "123",
346          "email": "user@site.com",
347          "ip_address": "2.125.160.216",
348          "username": "user",
349          "geo": {
350            "country_code": "GB",
351            "city": "Boxford",
352            "subdivision": "England",
353            "region": "United Kingdom"
354          }
355        }
356        "###);
357    }
358
359    #[test]
360    fn test_loose_type_requirements() {
361        let payload = include_str!("../../tests/fixtures/replay_failure_22_08_31.json");
362
363        let mut replay: Annotated<Replay> = Annotated::from_json(payload).unwrap();
364        normalize(
365            &mut replay,
366            None,
367            &RawUserAgentInfo::default(),
368            &GeoIpLookup::empty(),
369        );
370
371        let user = get_value!(replay.user!);
372        assert_eq!(user.ip_address.as_str(), Some("127.1.1.1"));
373        assert_eq!(user.username.value(), None);
374        assert_eq!(user.email.as_str(), Some("email@sentry.io"));
375        assert_eq!(user.id.as_str(), Some("1"));
376    }
377
378    #[test]
379    fn test_capped_values() {
380        let urls: Vec<Annotated<String>> = (0..101)
381            .map(|_| Annotated::new("localhost:9000".to_owned()))
382            .collect();
383
384        let error_ids: Vec<Annotated<Uuid>> = (0..101)
385            .map(|_| Annotated::new(Uuid::parse_str("52df9022835246eeb317dbd739ccd059").unwrap()))
386            .collect();
387
388        let trace_ids: Vec<Annotated<Uuid>> = (0..101)
389            .map(|_| Annotated::new(Uuid::parse_str("52df9022835246eeb317dbd739ccd059").unwrap()))
390            .collect();
391
392        let segment_names: Vec<Annotated<String>> = (0..101)
393            .map(|_| Annotated::new("/users/{id}".to_owned()))
394            .collect();
395
396        let mut replay = Annotated::new(Replay {
397            urls: Annotated::new(urls),
398            error_ids: Annotated::new(error_ids),
399            trace_ids: Annotated::new(trace_ids),
400            segment_names: Annotated::new(segment_names),
401            ..Default::default()
402        });
403
404        let replay_value = replay.value_mut().as_mut().unwrap();
405        normalize_array_fields(replay_value);
406
407        assert!(replay_value.error_ids.value().unwrap().len() == 100);
408        assert!(replay_value.trace_ids.value().unwrap().len() == 100);
409        assert!(replay_value.urls.value().unwrap().len() == 100);
410        assert!(replay_value.segment_names.value().unwrap().len() == 100);
411    }
412
413    #[test]
414    fn test_truncated_list_less_than_limit() {
415        let mut replay = Annotated::new(Replay {
416            urls: Annotated::new(Vec::new()),
417            error_ids: Annotated::new(Vec::new()),
418            trace_ids: Annotated::new(Vec::new()),
419            ..Default::default()
420        });
421
422        let replay_value = replay.value_mut().as_mut().unwrap();
423        normalize_array_fields(replay_value);
424
425        assert!(replay_value.error_ids.value().unwrap().is_empty());
426        assert!(replay_value.trace_ids.value().unwrap().is_empty());
427        assert!(replay_value.urls.value().unwrap().is_empty());
428    }
429
430    #[test]
431    fn test_error_id_validation() {
432        // NOTE: Interfaces will be tested separately.
433        let json = r#"{
434  "event_id": "52df9022835246eeb317dbd739ccd059",
435  "replay_id": "52df9022835246eeb317dbd739ccd059",
436  "segment_id": 0,
437  "replay_type": "session",
438  "error_sample_rate": 0.5,
439  "session_sample_rate": 0.5,
440  "timestamp": 946684800.0,
441  "replay_start_timestamp": 946684800.0,
442  "urls": ["localhost:9000"],
443  "error_ids": ["test"],
444  "trace_ids": [],
445  "platform": "myplatform",
446  "release": "myrelease",
447  "dist": "mydist",
448  "environment": "myenv",
449  "tags": [
450    [
451      "tag",
452      "value"
453    ]
454  ]
455}"#;
456
457        let mut replay = Annotated::<Replay>::from_json(json).unwrap();
458        let validation_result = validate(replay.value_mut().as_mut().unwrap());
459        assert!(validation_result.is_err());
460    }
461
462    #[test]
463    fn test_trace_id_validation() {
464        // NOTE: Interfaces will be tested separately.
465        let json = r#"{
466  "event_id": "52df9022835246eeb317dbd739ccd059",
467  "replay_id": "52df9022835246eeb317dbd739ccd059",
468  "segment_id": 0,
469  "replay_type": "session",
470  "error_sample_rate": 0.5,
471  "session_sample_rate": 0.5,
472  "timestamp": 946684800.0,
473  "replay_start_timestamp": 946684800.0,
474  "urls": ["localhost:9000"],
475  "error_ids": [],
476  "trace_ids": ["123"],
477  "platform": "myplatform",
478  "release": "myrelease",
479  "dist": "mydist",
480  "environment": "myenv",
481  "tags": [
482    [
483      "tag",
484      "value"
485    ]
486  ]
487}"#;
488
489        let mut replay = Annotated::<Replay>::from_json(json).unwrap();
490        let validation_result = validate(replay.value_mut().as_mut().unwrap());
491        assert!(validation_result.is_err());
492    }
493
494    #[test]
495    fn test_maxchars_trimming() {
496        let json = format!(r#"{{"dist": "{}"}}"#, "0".repeat(100));
497        let mut replay = Annotated::<Replay>::from_json(json.as_str()).unwrap();
498
499        normalize(
500            &mut replay,
501            None,
502            &RawUserAgentInfo::default(),
503            &GeoIpLookup::empty(),
504        );
505        assert_annotated_snapshot!(replay, @r###"
506        {
507          "platform": "other",
508          "dist": "0000000000000000000000000000000000000000000000000000000000000...",
509          "type": "replay_event",
510          "_meta": {
511            "dist": {
512              "": {
513                "rem": [
514                  [
515                    "!limit",
516                    "s",
517                    61,
518                    64
519                  ]
520                ],
521                "len": 100
522              }
523            }
524          }
525        }
526        "###);
527    }
528
529    #[test]
530    fn test_validate_u16_segment_id() {
531        // Does not fit within a u16.
532        let replay_id =
533            Annotated::new(EventId("52df9022835246eeb317dbd739ccd059".parse().unwrap()));
534        let segment_id: Annotated<u64> = Annotated::new(u16::MAX as u64 + 1);
535        let mut replay = Annotated::new(Replay {
536            replay_id,
537            segment_id,
538            ..Default::default()
539        });
540        assert!(validate(replay.value_mut().as_mut().unwrap()).is_err());
541
542        // Fits within a u16.
543        let replay_id =
544            Annotated::new(EventId("52df9022835246eeb317dbd739ccd059".parse().unwrap()));
545        let segment_id: Annotated<u64> = Annotated::new(u16::MAX as u64);
546        let mut replay = Annotated::new(Replay {
547            replay_id,
548            segment_id,
549            ..Default::default()
550        });
551        assert!(validate(replay.value_mut().as_mut().unwrap()).is_ok());
552    }
553}
relay_event_normalization/replay.rs

relay_event_normalization/
replay.rs
