diff --git a/Cargo.lock b/Cargo.lock index 1b46c5505d8..4a90046fbc4 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1723,6 +1723,12 @@ version = "0.32.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e629b9b98ef3dd8afe6ca2bd0f89306cec16d43d907889945bc5d6687f2f13c7" +[[package]] +name = "glob" +version = "0.3.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e4eba85ea1d0a966a983acd07deee566e67395d2d96b6fb39e62b5a833f1eb0b" + [[package]] name = "globset" version = "0.4.18" @@ -2505,6 +2511,12 @@ dependencies = [ "spin", ] +[[package]] +name = "leb128" +version = "0.2.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c83bff1d572d6b9aeef67ddfc8448e4a3737909cb28e81f97c791b9018703e52" + [[package]] name = "libbz2-rs-sys" version = "0.2.5" @@ -4986,8 +4998,20 @@ dependencies = [ name = "relay-serialization" version = "26.7.2" dependencies = [ + "leb128", + "prost 0.14.4", + "relay-serialization-derive", "serde", "serde_json", + "trybuild", +] + +[[package]] +name = "relay-serialization-derive" +version = "26.7.2" +dependencies = [ + "quote", + "syn", ] [[package]] @@ -5060,6 +5084,7 @@ dependencies = [ "relay-redis", "relay-replays", "relay-sampling", + "relay-serialization", "relay-spans", "relay-statsd", "relay-system", @@ -5794,6 +5819,15 @@ dependencies = [ "serde_core", ] +[[package]] +name = "serde_spanned" +version = "1.1.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6662b5879511e06e8999a8a235d848113e942c9124f211511b16466ee2995f26" +dependencies = [ + "serde_core", +] + [[package]] name = "serde_test" version = "1.0.177" @@ -6369,6 +6403,12 @@ version = "0.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7b2093cf4c8eb1e67749a6762251bc9cd836b6fc171623bd0a9d324d37af2417" +[[package]] +name = "target-triple" +version = "1.0.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c3a6bfce3d99adfa72d24750a61f782f3036a81e7f86d8841ee1326deaebd171" + [[package]] name = "tempfile" version = "3.27.0" @@ -6382,6 +6422,15 @@ dependencies = [ "windows-sys 0.61.2", ] +[[package]] +name = "termcolor" +version = "1.4.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "06794f8f6c5c898b3275aebefa6b8a1cb24cd2c6c79397ab15774837a0bc5755" +dependencies = [ + "winapi-util", +] + [[package]] name = "terminal_size" version = "0.4.4" @@ -6577,6 +6626,21 @@ dependencies = [ "tokio", ] +[[package]] +name = "toml" +version = "1.1.4+spec-1.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3aace63f4bbcdfc2c965b059de67119c89c4017a70d633be6c104910f67056f5" +dependencies = [ + "indexmap", + "serde_core", + "serde_spanned", + "toml_datetime", + "toml_parser", + "toml_writer", + "winnow", +] + [[package]] name = "toml_datetime" version = "1.1.1+spec-1.1.0" @@ -6600,13 +6664,19 @@ dependencies = [ [[package]] name = "toml_parser" -version = "1.1.2+spec-1.1.0" +version = "1.1.3+spec-1.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a2abe9b86193656635d2411dc43050282ca48aa31c2451210f4202550afb7526" +checksum = "1d38ac1cf9b95face32296c0a3ede1fdc270627c9d9c02a7274dd6d960dc4d56" dependencies = [ "winnow", ] +[[package]] +name = "toml_writer" +version = "1.1.2+spec-1.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7d56353a2a665ad0f41a421187180aab746c8c325620617ad883a99a1cbe66d2" + [[package]] name = "tonic" version = "0.14.6" @@ -6784,6 +6854,21 @@ version = "0.2.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e421abadd41a4225275504ea4d6566923418b7f05506fbc9c0fe86ba7396114b" +[[package]] +name = "trybuild" +version = "1.0.120" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1e605bf6b39357663d8ba4e984f8be8da8df6bb32e81031d6889024ea8fd68e4" +dependencies = [ + "glob", + "serde", + "serde_derive", + "serde_json", + "target-triple", + "termcolor", + "toml", +] + [[package]] name = "twox-hash" version = "2.1.2" diff --git a/Cargo.toml b/Cargo.toml index 5ca51c5fce8..a2305b17979 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -69,6 +69,7 @@ relay-redis = { path = "relay-redis" } relay-replays = { path = "relay-replays" } relay-sampling = { path = "relay-sampling" } relay-serialization = { path = "relay-serialization" } +relay-serialization-derive = { path = "relay-serialization-derive" } relay-server = { path = "relay-server" } relay-spans = { path = "relay-spans" } relay-statsd = { path = "relay-statsd" } @@ -145,6 +146,7 @@ insta = { version = "1", features = ["json", "redactions", "ron"] } ipnetwork = "0.21" itertools = "0.14" json-forensics = "0.1" +leb128 = "0.2.7" libc = "0.2" liblzma = "0.4" lru = "0.16" @@ -238,6 +240,7 @@ tower = { version = "0.5", default-features = false } tower-http = { version = "0.6", default-features = false } tracing = "0.1" tracing-subscriber = "0.3" +trybuild = "1.0.120" uaparser = "0.6" unescaper = "0.1" unicase = "2" diff --git a/relay-serialization-derive/Cargo.toml b/relay-serialization-derive/Cargo.toml new file mode 100644 index 00000000000..c628843533c --- /dev/null +++ b/relay-serialization-derive/Cargo.toml @@ -0,0 +1,20 @@ +[package] +name = "relay-serialization-derive" +authors = ["Sentry "] +description = "Derive macros for relay serialization tools" +homepage = "https://getsentry.github.io/relay/" +repository = "https://github.com/getsentry/relay" +version = "26.7.2" +edition = "2024" +license-file = "../LICENSE.md" +publish = false + +[lints] +workspace = true + +[dependencies] +quote = { workspace = true } +syn = { workspace = true } + +[lib] +proc-macro = true diff --git a/relay-serialization-derive/src/lib.rs b/relay-serialization-derive/src/lib.rs new file mode 100644 index 00000000000..6d5601dc926 --- /dev/null +++ b/relay-serialization-derive/src/lib.rs @@ -0,0 +1,177 @@ +//! A macro to add deserialization bounds to a prost message. +use proc_macro::TokenStream; +use quote::{quote, quote_spanned}; +use syn::spanned::Spanned; +use syn::{DeriveInput, LitStr, PathArguments, Type, parse_macro_input}; + +#[proc_macro_derive(RuntimeDescription)] +pub fn derive(s: TokenStream) -> TokenStream { + let input = parse_macro_input!(s as DeriveInput); + + let mut nested_v = vec![]; + + match input.data { + syn::Data::Struct(data_struct) => { + if let syn::Fields::Named(fields_named) = data_struct.fields { + for field in fields_named.named.iter() { + for attr in &field.attrs { + if attr.path().is_ident("prost") + && let Err(value) = collect_tags_and_types(&mut nested_v, field, attr) + { + return value; + } + } + } + } + } + syn::Data::Enum(data_enum) => { + for variant in data_enum.variants.iter() { + for attr in &variant.attrs { + if attr.path().is_ident("prost") + && let Some(first_field) = variant.fields.iter().next() + && let Err(value) = collect_tags_and_types(&mut nested_v, first_field, attr) + { + return value; + } + } + } + } + syn::Data::Union(_) => { + let span = input.span(); + return quote_spanned! { + span => compile_error!("unions are unsupported") + } + .into(); + } + } + + let qs = nested_v.into_iter().map(|(typ, tag)| { + if let TypeKind::Field(tag) = tag { + quote! { + ::relay_serialization::prost::Nested::Field(#tag, < #typ as ::relay_serialization::prost::RuntimeDescription>::desc) + } + } else { + quote! { + ::relay_serialization::prost::Nested::Oneof(< #typ as ::relay_serialization::prost::RuntimeDescription>::desc) + } + } + }); + + let typ = &input.ident; + quote! { + impl ::relay_serialization::prost::RuntimeDescription for #typ { + fn desc() -> &'static [::relay_serialization::prost::Nested] { + &[#(#qs,)*] + } + } + } + .into() +} + +enum TypeKind { + Field(u32), + OneOf, +} + +fn collect_tags_and_types( + tags_and_types: &mut Vec<(Type, TypeKind)>, + field: &syn::Field, + attr: &syn::Attribute, +) -> Result<(), TokenStream> { + let mut tag: Option = None; + let mut message_type: Option<&Type> = None; + + let result = attr.parse_nested_meta(|meta| { + // Fun landmine: parse_nested_meta assumes you consume all the parsed--for tags like + // "#[foo(bar)]", bar is consumed automatically just before this callback is called, + // but "#[foo(bar = 6)]", the "= 6" value is not, and you MUST consume that in this callback + // or the parser will try to parse the "next" value in the stream, but the stream + // won't have advanced correctly, leading to sadness. + if meta.path.is_ident("tag") { + let value = meta.value()?; + let s: LitStr = value.parse()?; + let tag_val = s + .value() + .parse::() + .map_err(|_| meta.input.error("error parsing tag"))?; + + tag = TypeKind::Field(tag_val).into(); + return Ok(()); + } + + if meta.path.is_ident("map") + || meta.path.is_ident("btree_map") + || meta.path.is_ident("hash_map") + { + return Err(meta.input.error("map types are currently unsupported")); + } + + if meta.path.is_ident("group") { + return Err(meta.input.error("group types are unsupported")); + } + + if meta.path.is_ident("message") { + message_type = innermost_contained_type(&field.ty); + return Ok(()); + } + + if meta.path.is_ident("oneof") { + message_type = innermost_contained_type(&field.ty); + let value = meta.value()?; + let _: LitStr = value.parse()?; // Consumes the tags + + // Tag = None communicates that this is a one-of. + tag = Some(TypeKind::OneOf); + return Ok(()); + } + + // Things we don't care about, but are more than just a tag + if meta.path.is_ident("bytes") + || meta.path.is_ident("enumeration") + || meta.path.is_ident("tags") + || meta.path.is_ident("packed") + { + let value = meta.value()?; + let _: LitStr = value.parse()?; + return Ok(()); + } + + // By default, just skip everything else; those should be just tags (and if not, we'll get + // an error, albeit a confusing one.) + Ok(()) + }); + + if let Err(e) = result { + return Err(e.to_compile_error().into()); + } + + if let Some(typ) = message_type { + let Some(tag) = tag else { + let span = attr.span(); + return Err(quote_spanned! { + span => compile_error!("missing tag value"); + } + .into()); + }; + tags_and_types.push((typ.to_owned(), tag)); + } + Ok(()) +} + +fn innermost_contained_type(typ: &Type) -> Option<&Type> { + let Type::Path(p) = typ else { + return None; + }; + + let last_part = p.path.segments.last()?; + + if let PathArguments::AngleBracketed(inner) = &last_part.arguments { + let first = inner.args.first()?; + match first { + syn::GenericArgument::Type(t) => innermost_contained_type(t), + _ => None, + } + } else { + Some(typ) + } +} diff --git a/relay-serialization/Cargo.toml b/relay-serialization/Cargo.toml index acef4aaf4a7..06971776c5c 100644 --- a/relay-serialization/Cargo.toml +++ b/relay-serialization/Cargo.toml @@ -13,7 +13,11 @@ publish = false workspace = true [dependencies] +prost = { workspace = true } serde = { workspace = true } +leb128 = { workspace = true } [dev-dependencies] serde_json = { workspace = true } +relay-serialization-derive = { workspace = true } +trybuild = { workspace = true } diff --git a/relay-serialization/src/lib.rs b/relay-serialization/src/lib.rs index c7b71d62048..9ac3f7eeb32 100644 --- a/relay-serialization/src/lib.rs +++ b/relay-serialization/src/lib.rs @@ -1,7 +1,63 @@ //! Relay serialization toolkit. //! //! Used to ensure we can bound the work when deserializing untrusted data. - +#![recursion_limit = "256"] #![warn(missing_docs)] +use std::fmt; + +/// A budget for the ops a single deserialization is allowed to spend. +struct Meter { + limit: usize, + remaining: usize, + exceeded: bool, +} + +impl Meter { + /// Creates a meter which allows spending at most `limit` operations. + pub fn new(limit: usize) -> Self { + Self { + limit, + remaining: limit, + exceeded: false, + } + } + + /// Returns the number of ops spent. + pub fn spent(&self) -> usize { + self.limit - self.remaining + } + + /// Returns true if we've exceeded our budget. + pub fn exceeded(&self) -> bool { + self.exceeded + } + + /// Tries to charge `amount` operations to the budget. If we exceed, we return an error, + /// set the remaining budget to 0, and mark the budget as exceeded. + pub fn spend(&mut self, amount: usize) -> Result<(), LimitExceeded> { + match self.remaining.checked_sub(amount) { + Some(remaining) => { + self.remaining = remaining; + Ok(()) + } + None => { + self.remaining = 0; + self.exceeded = true; + Err(LimitExceeded) + } + } + } +} + +/// The error produced when a [`Meter`] runs out of budget. +struct LimitExceeded; + +impl fmt::Display for LimitExceeded { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + write!(f, "deserialization exceeds the operation budget") + } +} + +pub mod prost; pub mod serde; diff --git a/relay-serialization/src/prost/mod.rs b/relay-serialization/src/prost/mod.rs new file mode 100644 index 00000000000..9b74cfdc98a --- /dev/null +++ b/relay-serialization/src/prost/mod.rs @@ -0,0 +1,11 @@ +//! Deserialization routines for prost. +//! This implements a scanner to record the number of "operations" needed to decode a proto, +//! allowing a caller to enforce a hard limit on how much work to be done. +mod scan; + +pub use scan::Error; +pub use scan::decode; +pub use scan::scan; +pub use scan::{Nested, RuntimeDescription}; + +pub use scan::ops; diff --git a/relay-serialization/src/prost/scan.rs b/relay-serialization/src/prost/scan.rs new file mode 100644 index 00000000000..db26d87a3f5 --- /dev/null +++ b/relay-serialization/src/prost/scan.rs @@ -0,0 +1,264 @@ +use prost::{DecodeError, Message}; +use std::fmt; + +use crate::{LimitExceeded, Meter}; + +/// Costs associated with different kinds of operations; right now, just one cost for every field +/// occurrence on the wire (but leave the door open for more.) +mod cost { + pub const FIELD: usize = 1; +} + +/// The maximum nesting depth the scanner walks before giving up. This matches prost's own +/// `RECURSION_LIMIT`. +const RECURSION_LIMIT: u32 = 100; + +/// Protobuf wire types, as encoded in the bottom three bits of a field key. +mod wire_type { + pub const VARINT: u8 = 0; + pub const SIXTY_FOUR_BIT: u8 = 1; + pub const LENGTH_DELIMITED: u8 = 2; + pub const START_GROUP: u8 = 3; + pub const END_GROUP: u8 = 4; + pub const THIRTY_TWO_BIT: u8 = 5; +} + +/// The kind of nested field encountered in the proto. Either a regular tagged field, or +/// a "oneof" which will have its tags on the actual nested message, +pub enum Nested { + /// A regular nested message field. + Field(u32, fn() -> &'static [Nested]), + + /// A oneof nested field. + Oneof(fn() -> &'static [Nested]), +} + +/// A trait for prost Messages to implement, allowing them to self-describe the kinds of nested +/// fields that have on themselves. +pub trait RuntimeDescription { + /// Returns the one and only array of nested fields (fields that refer to other messages) + /// on this particular Message. + fn desc() -> &'static [Nested]; +} + +/// An error returned by [`scan`] or [`decode`]. +#[derive(Debug)] +pub enum Error { + /// A scanner error (reading a primitive, reaching a recursion limit, etc.) + ScanError(String), + /// Actually hitting the op budget for the decode. + LimitExceeded, + /// A decoder error, originating from prost. + Decode(DecodeError), +} + +impl Error { + /// Returns `true` if the message decoding exceeded the budget. + pub fn is_limit_exceeded(&self) -> bool { + matches!(self, Self::LimitExceeded) + } + + /// Returns `true` if the message decoding encountered a decoding (prost) error. + pub fn is_decode_error(&self) -> bool { + matches!(self, Self::Decode(_)) + } + + /// Returns `true` if the message decoding encountered a parsing error during scanning. + pub fn is_scan_error(&self) -> bool { + matches!(self, Self::ScanError(_)) + } +} + +impl fmt::Display for Error { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + Self::ScanError(msg) => write!(f, "scanner error: {}", msg), + Self::LimitExceeded => write!(f, "message exceeds the operation limit"), + Self::Decode(error) => error.fmt(f), + } + } +} + +impl std::error::Error for Error { + fn source(&self) -> Option<&(dyn std::error::Error + 'static)> { + match self { + Self::ScanError(_) => None, + Self::LimitExceeded => None, + Self::Decode(error) => Some(error), + } + } +} + +/// Decodes an `M` from `buf`, spending at most `max_ops` doing so. Returns +/// [`Error::LimitExceeded`] if the message exceeds the budget. +pub fn decode(buf: &[u8], max_ops: usize) -> Result +where + M: Message + Default + RuntimeDescription, +{ + scan::(buf, max_ops)?; + M::decode(buf).map_err(Error::Decode) +} + +/// Checks that the message in `buf` fits within `max_ops`, without decoding it. Returns +/// the number of ops spent doing the scan. +pub fn scan(buf: &[u8], max_ops: usize) -> Result { + let mut meter = Meter::new(max_ops); + + match scan_message(buf, T::desc(), &mut meter, 0) { + Ok(()) => Ok(meter.spent()), + // The budget is checked first, because it travels as an ordinary decode error. + Err(_) if meter.exceeded() => Err(Error::LimitExceeded), + Err(error) => Err(error), + } +} + +impl From for Error { + fn from(_: LimitExceeded) -> Self { + Error::LimitExceeded + } +} +fn find(table: &'static [Nested], tag: u32, limit: u32) -> Option<&'static [Nested]> { + if limit == 0 { + return None; + } + table.iter().find_map(|entry| match entry { + Nested::Field(t, desc) if *t == tag => Some(desc()), + Nested::Oneof(group) => find(group(), tag, limit - 1), + _ => None, + }) +} + +fn scan_message( + buf: &[u8], + desc: &'static [Nested], + meter: &mut Meter, + depth: u32, +) -> Result<(), Error> { + if depth > RECURSION_LIMIT { + return Err(Error::ScanError("recursion limit reached".to_owned())); + } + + let mut reader = Reader(buf); + while !reader.is_empty() { + let (tag, wire_type) = key(&mut reader)?; + meter.spend(cost::FIELD)?; + scan_field(&mut reader, tag, wire_type, Some(desc), meter, depth)?; + } + + Ok(()) +} + +/// Consumes the body of a single field, recursing if the schema says it holds a message. +fn scan_field( + reader: &mut Reader<'_>, + tag: u32, + wire_type: u8, + desc: Option<&'static [Nested]>, + meter: &mut Meter, + depth: u32, +) -> Result<(), Error> { + match wire_type { + wire_type::VARINT => { + reader.read_varint()?; + } + wire_type::SIXTY_FOUR_BIT => { + reader.read_exact(8)?; + } + wire_type::THIRTY_TWO_BIT => { + reader.read_exact(4)?; + } + wire_type::LENGTH_DELIMITED => { + let len: usize = usize::try_from(reader.read_varint()?) + .map_err(|_| Error::ScanError("buffer underflow".to_owned()))?; + let payload = reader.read_exact(len)?; + + // Only recurse where our generated schema tells us we have a nested message. This is + // how we distinguish between strings/repeated bytes, and genuine nested messages. + if let Some(nested) = desc.and_then(|desc| find(desc, tag, RECURSION_LIMIT)) { + scan_message(payload, nested, meter, depth + 1)?; + } + } + // proto3 has no groups, so nothing inside one can have a descriptor. This exists so that a + // payload carrying group wire types is walked rather than rejected, matching what prost's + // `skip_field` accepts. + wire_type::START_GROUP => scan_group(reader, tag, meter, depth + 1)?, + wire_type::END_GROUP => { + return Err(Error::ScanError("unexpected end group tag".to_owned())); + } + _ => return Err(Error::ScanError("invalid wire type value".to_owned())), + } + + Ok(()) +} + +fn scan_group( + reader: &mut Reader<'_>, + group_tag: u32, + meter: &mut Meter, + depth: u32, +) -> Result<(), Error> { + if depth > RECURSION_LIMIT { + return Err(Error::ScanError("recursion limit reached".to_owned())); + } + + loop { + let (tag, wire_type) = key(reader)?; + meter.spend(cost::FIELD)?; + + if wire_type == wire_type::END_GROUP { + if tag != group_tag { + return Err(Error::ScanError("unexpected end group tag".to_owned())); + } + return Ok(()); + } + + scan_field(reader, tag, wire_type, None, meter, depth)?; + } +} + +fn key(reader: &mut Reader<'_>) -> Result<(u32, u8), Error> { + let key = reader.read_varint()?; + let wire_type = (key & 0b111) as u8; + let tag = + u32::try_from(key >> 3).map_err(|_| Error::ScanError("invalid tag value".to_owned()))?; + + if tag == 0 { + return Err(Error::ScanError("invalid tag value".to_owned())); + } + + Ok((tag, wire_type)) +} + +// A little wrapper to assist with reading and consuming bytes from a proto byte-buffer. +struct Reader<'a>(&'a [u8]); + +impl<'a> Reader<'a> { + fn is_empty(&self) -> bool { + self.0.is_empty() + } + + fn read_varint(&mut self) -> Result { + leb128::read::unsigned(&mut self.0) + .map_err(|_| Error::ScanError("invalid varint".to_owned())) + } + + fn read_exact(&mut self, len: usize) -> Result<&'a [u8], Error> { + if len > self.0.len() { + return Err(Error::ScanError("buffer underflow".to_owned())); + } + + let (payload, rest) = self.0.split_at(len); + self.0 = rest; + Ok(payload) + } +} + +/// Returns the number of ops consumed deserializing the supplied message. +pub fn ops(msg: &T) -> usize +where + T: RuntimeDescription + Message, +{ + let mut meter = Meter::new(usize::MAX); + scan_message(&msg.encode_to_vec(), T::desc(), &mut meter, 0).unwrap(); + meter.spent() +} diff --git a/relay-serialization/src/serde/de.rs b/relay-serialization/src/serde/de.rs index ecf5dfed8bb..feb0ab9f3b9 100644 --- a/relay-serialization/src/serde/de.rs +++ b/relay-serialization/src/serde/de.rs @@ -5,73 +5,19 @@ use serde::de::{ use std::fmt; use std::marker::PhantomData; +use crate::Meter; + /// Costs associated with different kinds of operations; right now, just have one cost for /// all operations (but leave the door open for more.) mod cost { pub const UNIT: usize = 1; } -/// A budget for the ops a single deserialization is allowed to spend. -struct Meter { - #[cfg(test)] - limit: usize, - remaining: usize, - exceeded: bool, -} - impl Meter { - /// Creates a meter which allows spending at most `limit` operations. - pub fn new(limit: usize) -> Self { - Self { - #[cfg(test)] - limit, - remaining: limit, - exceeded: false, - } - } - /// Wraps `deserializer`, so that everything it produces is charged to this meter. - pub fn wrap<'de, D: Deserializer<'de>>( - &mut self, - deserializer: D, - ) -> MeteredDeserializer<'_, D> { + fn wrap<'de, D: Deserializer<'de>>(&mut self, deserializer: D) -> MeteredDeserializer<'_, D> { MeteredDeserializer::new(self, deserializer) } - - #[cfg(test)] - fn spent(&self) -> usize { - self.limit - self.remaining - } - - /// Returns true if we've exceeded our budget. - pub fn exceeded(&self) -> bool { - self.exceeded - } - - /// Tries to charge `amount` operations to the budget. If we exceed, we return an error, - /// set the remaining budget to 0, and mark the budget as exceeded. - pub fn spend(&mut self, amount: usize) -> Result<(), E> { - match self.remaining.checked_sub(amount) { - Some(remaining) => { - self.remaining = remaining; - Ok(()) - } - None => { - self.remaining = 0; - self.exceeded = true; - Err(serde_de::Error::custom(LimitExceeded {})) - } - } - } -} - -/// The error produced when a [`Meter`] runs out of budget. -struct LimitExceeded(); - -impl fmt::Display for LimitExceeded { - fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { - write!(f, "deserialization exceeds the operation budget") - } } /// An error returned by [`deserialize`]. @@ -256,7 +202,7 @@ macro_rules! visit_scalar { ($($method:ident($ty:ty)),* $(,)?) => { $( fn $method(self, v: $ty) -> Result { - self.meter.spend(cost::UNIT)?; + self.meter.spend(cost::UNIT).map_err(E::custom)?; self.inner.$method(v) } )* @@ -288,42 +234,42 @@ impl<'de, V: Visitor<'de>> Visitor<'de> for MeteredVisitor<'_, V> { } fn visit_str(self, v: &str) -> Result { - self.meter.spend(cost::UNIT)?; + self.meter.spend(cost::UNIT).map_err(E::custom)?; self.inner.visit_str(v) } fn visit_borrowed_str(self, v: &'de str) -> Result { - self.meter.spend(cost::UNIT)?; + self.meter.spend(cost::UNIT).map_err(E::custom)?; self.inner.visit_borrowed_str(v) } fn visit_string(self, v: String) -> Result { - self.meter.spend(cost::UNIT)?; + self.meter.spend(cost::UNIT).map_err(E::custom)?; self.inner.visit_string(v) } fn visit_bytes(self, v: &[u8]) -> Result { - self.meter.spend(cost::UNIT)?; + self.meter.spend(cost::UNIT).map_err(E::custom)?; self.inner.visit_bytes(v) } fn visit_borrowed_bytes(self, v: &'de [u8]) -> Result { - self.meter.spend(cost::UNIT)?; + self.meter.spend(cost::UNIT).map_err(E::custom)?; self.inner.visit_borrowed_bytes(v) } fn visit_byte_buf(self, v: Vec) -> Result { - self.meter.spend(cost::UNIT)?; + self.meter.spend(cost::UNIT).map_err(E::custom)?; self.inner.visit_byte_buf(v) } fn visit_none(self) -> Result { - self.meter.spend(cost::UNIT)?; + self.meter.spend(cost::UNIT).map_err(E::custom)?; self.inner.visit_none() } fn visit_unit(self) -> Result { - self.meter.spend(cost::UNIT)?; + self.meter.spend(cost::UNIT).map_err(E::custom)?; self.inner.visit_unit() } @@ -340,7 +286,9 @@ impl<'de, V: Visitor<'de>> Visitor<'de> for MeteredVisitor<'_, V> { } fn visit_seq>(self, seq: A) -> Result { - self.meter.spend(cost::UNIT)?; + self.meter + .spend(cost::UNIT) + .map_err(serde_de::Error::custom)?; self.inner.visit_seq(MeteredSeqAccess { meter: self.meter, inner: seq, @@ -348,7 +296,9 @@ impl<'de, V: Visitor<'de>> Visitor<'de> for MeteredVisitor<'_, V> { } fn visit_map>(self, map: A) -> Result { - self.meter.spend(cost::UNIT)?; + self.meter + .spend(cost::UNIT) + .map_err(serde_de::Error::custom)?; self.inner.visit_map(MeteredMapAccess { meter: self.meter, @@ -465,7 +415,9 @@ impl<'de, A: VariantAccess<'de>> VariantAccess<'de> for MeteredVariantAccess<'_, type Error = A::Error; fn unit_variant(self) -> Result<(), Self::Error> { - self.meter.spend(cost::UNIT)?; + self.meter + .spend(cost::UNIT) + .map_err(serde_de::Error::custom)?; self.inner.unit_variant() } diff --git a/relay-serialization/tests/build_failures/map_fails.rs b/relay-serialization/tests/build_failures/map_fails.rs new file mode 100644 index 00000000000..97491db200b --- /dev/null +++ b/relay-serialization/tests/build_failures/map_fails.rs @@ -0,0 +1,24 @@ +use prost::Message; +use relay_serialization_derive::RuntimeDescription; +use std::collections::BTreeMap; +use std::collections::HashMap; + +#[derive(Clone, PartialEq, Message, RuntimeDescription)] +struct MapDoesNotCompile { + #[prost(map = "string, string", tag = "1")] + value: HashMap, +} + +#[derive(Clone, PartialEq, Message, RuntimeDescription)] +struct HashMapDoesNotCompile { + #[prost(hash_map = "string, string", tag = "1")] + value: HashMap, +} + +#[derive(Clone, PartialEq, Message, RuntimeDescription)] +struct BTreeMapDoesNotCompile { + #[prost(btree_map = "string, string", tag = "1")] + value: BTreeMap, +} + +fn main() {} diff --git a/relay-serialization/tests/build_failures/map_fails.stderr b/relay-serialization/tests/build_failures/map_fails.stderr new file mode 100644 index 00000000000..41746f130c0 --- /dev/null +++ b/relay-serialization/tests/build_failures/map_fails.stderr @@ -0,0 +1,17 @@ +error: map types are currently unsupported + --> tests/build_failures/map_fails.rs:8:17 + | +8 | #[prost(map = "string, string", tag = "1")] + | ^ + +error: map types are currently unsupported + --> tests/build_failures/map_fails.rs:14:22 + | +14 | #[prost(hash_map = "string, string", tag = "1")] + | ^ + +error: map types are currently unsupported + --> tests/build_failures/map_fails.rs:20:23 + | +20 | #[prost(btree_map = "string, string", tag = "1")] + | ^ diff --git a/relay-serialization/tests/prost_tests.rs b/relay-serialization/tests/prost_tests.rs new file mode 100644 index 00000000000..a14c84bc06e --- /dev/null +++ b/relay-serialization/tests/prost_tests.rs @@ -0,0 +1,200 @@ +mod prost { + use prost::Message; + use relay_serialization::prost::{decode, ops, scan}; + use relay_serialization_derive::RuntimeDescription as DeriveRuntimeDescription; + + /// A leaf message with one field of every shape the scanner dispatches on. + #[derive(Clone, PartialEq, Message, DeriveRuntimeDescription)] + struct Leaf { + #[prost(string, tag = "1")] + text: String, + #[prost(uint64, tag = "2")] + number: u64, + #[prost(double, tag = "3")] + double: f64, + #[prost(fixed32, tag = "4")] + fixed: u32, + #[prost(bytes = "vec", tag = "5")] + blob: Vec, + #[prost(oneof = "Var", tags = "6, 7")] + one_of: ::core::option::Option, + } + + /// A message which nests, so the scanner has to recurse to see the leaves. + #[derive(Clone, PartialEq, Message, DeriveRuntimeDescription)] + struct Branch { + #[prost(message, repeated, tag = "1")] + leaves: Vec, + #[prost(message, optional, boxed, tag = "2")] + branch: Option>, + #[prost(string, tag = "3")] + label: String, + } + + /// A message which nests, so the scanner has to recurse to see the leaves. + #[derive(Clone, PartialEq, prost::Oneof, DeriveRuntimeDescription)] + enum Var { + #[prost(string, tag = "6")] + Name(String), + // /// Recursion through a `oneof`. + #[prost(message, tag = "7")] + Leaf(Box), + } + + #[test] + fn test_scan_charges_each_empty_element() { + // The case a size bound cannot see: every element is tiny, but there are a great many. + for count in [0, 1, 2, 512] { + let branch = Branch { + leaves: vec![Leaf::default(); count], + ..Default::default() + }; + + // One op for each `leaves` occurrence. An empty `Leaf` has no fields of its own, and + // default scalars are not encoded at all in proto3. + assert_eq!(ops(&branch), count); + } + } + + #[test] + fn test_scan_charges_opaque_payload_once() { + // A megabyte in one field costs one op, where a byte meter would charge a megabyte. + let branch = Branch { + label: "a".repeat(1 << 20), + ..Default::default() + }; + + assert_eq!(ops(&branch), 1); + } + + #[test] + fn test_scan_charges_every_depth() { + let branch = Branch { + leaves: vec![ + Leaf { + text: "one".to_owned(), + number: 1, + double: 1.5, + fixed: 2, + blob: vec![1, 2, 3], + one_of: Some(Var::Leaf(Box::new(Branch { + leaves: vec![], + branch: None, + label: "done".to_owned(), + }))), + }, + Leaf { + number: 7, + ..Default::default() + }, + ], + branch: Some(Box::new(Branch { + label: "inner".to_owned(), + ..Default::default() + })), + label: "outer".to_owned(), + }; + + // Two `leaves` occurrences carrying five and one field, one `branch` carrying one field, + // the outer `label`, and then finally the one_of field + the child branch's label. + let expected = (1 + 5) + (1 + 1) + (1 + 1) + 1 + (1 + 1); + assert_eq!(ops(&branch), expected); + } + + #[test] + fn test_scan_exceeds_budget() { + let branch = Branch { + leaves: vec![Leaf::default(); 4096], + ..Default::default() + }; + + let buf = branch.encode_to_vec(); + let error = scan::(&buf, 128).unwrap_err(); + assert!(error.is_limit_exceeded()); + assert_eq!(error.to_string(), "message exceeds the operation limit"); + } + + #[test] + fn test_scan_exceeds_budget_when_nested() { + // The budget has to survive recursion: the fields are all four levels down. + let leaves = vec![Leaf::default(); 4096]; + let branch = Branch { + branch: Some(Box::new(Branch { + branch: Some(Box::new(Branch { + leaves, + ..Default::default() + })), + ..Default::default() + })), + ..Default::default() + }; + + let buf = branch.encode_to_vec(); + let error = scan::(&buf, 256).unwrap_err(); + assert!(error.is_limit_exceeded()); + } + + #[test] + fn test_decode_rejects_before_decoding() { + let branch = Branch { + leaves: vec![Leaf::default(); 4096], + ..Default::default() + }; + let payload = branch.encode_to_vec(); + + assert!( + decode::(&payload, 128) + .unwrap_err() + .is_limit_exceeded() + ); + // The same payload decodes once the budget accommodates it. + let decoded = decode::(&payload, 1 << 20).unwrap(); + assert_eq!(decoded, branch); + } + + #[test] + fn test_scan_accepts_what_prost_accepts() { + // An unknown field prost would skip: tag 9, length delimited, holding a nested message the + // scanner has no descriptor for. It is charged once and walked over. + let payload = [0x4a, 0x04, 0x08, 0x01, 0x10, 0x02]; + + assert!(Branch::decode(payload.as_slice()).is_ok()); + assert!(scan::(&payload, 1).is_ok()); + } + + #[test] + fn test_scan_rejects_malformed_payloads() { + // A truncated length delimiter, a tag of zero, and a varint with no terminator. + for payload in [ + [0x1a, 0x08, 0x61].as_slice(), + [0x00, 0x01].as_slice(), + [0x08, 0xff].as_slice(), + ] { + let error = decode::(payload, 1 << 20).unwrap_err(); + assert!(error.is_scan_error(), "{payload:?}"); + // Whatever the scanner rejects, prost rejects too. + assert!(Branch::decode(payload).is_err(), "{payload:?}"); + } + } + + #[test] + fn test_scan_bounds_its_own_recursion() { + // Deeper than the recursion limit, so the scanner must not run out of stack walking it. + let mut payload = Vec::new(); + for _ in 0..101 { + let mut framed = vec![0x12, payload.len() as u8]; + framed.extend_from_slice(&payload); + payload = framed; + } + + let error = scan::(&payload, 1 << 20).unwrap_err(); + assert!(!error.is_limit_exceeded()); + assert!(Branch::decode(payload.as_slice()).is_err()); + } + + #[test] + fn test_map_fails_build() { + let t = trybuild::TestCases::new(); + t.compile_fail("tests/build_failures/map_fails.rs"); + } +} diff --git a/relay-server/Cargo.toml b/relay-server/Cargo.toml index 036ff8a2618..6083b42e555 100644 --- a/relay-server/Cargo.toml +++ b/relay-server/Cargo.toml @@ -91,6 +91,7 @@ relay-protocol = { workspace = true } relay-quotas = { workspace = true } relay-redis = { workspace = true } relay-replays = { workspace = true } +relay-serialization = { workspace = true } relay-conventions = { workspace = true } relay-sampling = { workspace = true } relay-spans = { workspace = true } diff --git a/relay/src/main.rs b/relay/src/main.rs index d324b43ea74..68dca04c71a 100644 --- a/relay/src/main.rs +++ b/relay/src/main.rs @@ -59,6 +59,7 @@ //! - [`relay-replays`]: Session replay protocol and processing. //! - [`relay-sampling`]: Dynamic sampling functionality. //! - [`relay-serialization`]: Bounded deserialization functionality. +//! - [`relay-serialization-derive`]: Derive macros for relay deserialization functionality. //! - [`relay-server`]: Endpoints and services. //! - [`relay-statsd`]: High-level StatsD metric client for internal measurements. //! - [`relay-system`]: Foundational system components for Relay's services. @@ -102,6 +103,7 @@ //! [`relay-replays`]: ../relay_replays/index.html //! [`relay-sampling`]: ../relay_sampling/index.html //! [`relay-serialization`]: ../relay_serialization/index.html +//! [`relay-serialization-derive`]: ../relay_serialization_derive/index.html //! [`relay-server`]: ../relay_server/index.html //! [`relay-statsd`]: ../relay_statsd/index.html //! [`relay-system`]: ../relay_system/index.html