|
| 1 | +//! UTF-8-tolerant `/api/v2/series` `MetricPayload` decode. |
| 2 | +//! |
| 3 | +//! rust-protobuf validates UTF-8 on `string` fields and rejects the entire payload on |
| 4 | +//! any non-UTF-8 byte. The Datadog Agent forwards feral non-UTF-8 `DogStatsD` names and |
| 5 | +//! tags into those fields unchecked, so a strict `MetricPayload::parse_from_bytes` drops |
| 6 | +//! every legitimate agent-lane payload that carries one bad byte, along with all the |
| 7 | +//! contexts it holds. |
| 8 | +//! |
| 9 | +//! This mirrors the production intake, which retries a failed strict decode into a |
| 10 | +//! tags-as-`bytes` message and keeps the payload. So only the `tags` field tolerates |
| 11 | +//! non-UTF-8 here, lossy-converted. Every other string field stays UTF-8-strict, so a |
| 12 | +//! non-UTF-8 metric name still drops the payload exactly as production does. Genuinely |
| 13 | +//! malformed wire also fails. |
| 14 | +
|
| 15 | +use datadog_protos::metrics::metric_payload::{MetricPoint, MetricSeries, Resource}; |
| 16 | +use datadog_protos::metrics::MetricPayload; |
| 17 | +use protobuf::rt::WireType; |
| 18 | +use protobuf::{CodedInputStream, EnumOrUnknown}; |
| 19 | + |
| 20 | +/// Why the intake did not decode a `/api/v2/series` body. |
| 21 | +#[derive(Clone, Copy, Debug, Eq, PartialEq)] |
| 22 | +pub(crate) enum Rejection { |
| 23 | + /// A non-tag string field held non-UTF-8 bytes. Production rejects the payload too, so this |
| 24 | + /// is expected behavior on feral input, not a decode defect. |
| 25 | + NonUtf8StrictField, |
| 26 | + /// The bytes are not a well-formed protobuf message. No real producer should emit this. |
| 27 | + MalformedWire, |
| 28 | +} |
| 29 | + |
| 30 | +impl From<protobuf::Error> for Rejection { |
| 31 | + fn from(_: protobuf::Error) -> Self { |
| 32 | + Rejection::MalformedWire |
| 33 | + } |
| 34 | +} |
| 35 | + |
| 36 | +/// Replace each contiguous run of invalid UTF-8 bytes with a single U+FFFD, matching Go's |
| 37 | +/// `strings.ToValidUTF8` as production's `sanitizePayload` applies it to Agent tags. Rust's |
| 38 | +/// `from_utf8_lossy` emits one replacement per maximal subpart, which yields a different tag |
| 39 | +/// string for a multi-byte invalid run and so a false divergence. Only tags use this. |
| 40 | +fn to_valid_utf8(bytes: &[u8]) -> String { |
| 41 | + let mut out = String::with_capacity(bytes.len()); |
| 42 | + let mut prev_invalid = false; |
| 43 | + for chunk in bytes.utf8_chunks() { |
| 44 | + let valid = chunk.valid(); |
| 45 | + if !valid.is_empty() { |
| 46 | + out.push_str(valid); |
| 47 | + prev_invalid = false; |
| 48 | + } |
| 49 | + if !chunk.invalid().is_empty() { |
| 50 | + if !prev_invalid { |
| 51 | + out.push('\u{FFFD}'); |
| 52 | + } |
| 53 | + prev_invalid = true; |
| 54 | + } |
| 55 | + } |
| 56 | + out |
| 57 | +} |
| 58 | + |
| 59 | +/// Strict UTF-8. Production keeps every non-tag string field UTF-8-validated, so a |
| 60 | +/// non-UTF-8 byte here drops the payload as it would in production. |
| 61 | +fn strict(bytes: Vec<u8>) -> Result<String, Rejection> { |
| 62 | + String::from_utf8(bytes).map_err(|_| Rejection::NonUtf8StrictField) |
| 63 | +} |
| 64 | + |
| 65 | +fn unpack(tag: u32) -> Result<(u32, WireType), Rejection> { |
| 66 | + let wire = WireType::new(tag & 0x7).ok_or(Rejection::MalformedWire)?; |
| 67 | + Ok((tag >> 3, wire)) |
| 68 | +} |
| 69 | + |
| 70 | +/// Decode a `/api/v2/series` body into a `MetricPayload`. Only `tags` tolerate non-UTF-8, |
| 71 | +/// lossy-converted. A non-UTF-8 non-tag string field returns `NonUtf8StrictField`, matching |
| 72 | +/// production's reject. Structurally malformed protobuf returns `MalformedWire`. |
| 73 | +pub(crate) fn decode_metric_payload(body: &[u8]) -> Result<MetricPayload, Rejection> { |
| 74 | + let mut is = CodedInputStream::from_bytes(body); |
| 75 | + let mut payload = MetricPayload::new(); |
| 76 | + while let Some(tag) = is.read_raw_tag_or_eof()? { |
| 77 | + match unpack(tag)? { |
| 78 | + (1, WireType::LengthDelimited) => payload.series.push(decode_series(&is.read_bytes()?)?), |
| 79 | + (_, wire) => is.skip_field(wire)?, |
| 80 | + } |
| 81 | + } |
| 82 | + Ok(payload) |
| 83 | +} |
| 84 | + |
| 85 | +fn decode_series(body: &[u8]) -> Result<MetricSeries, Rejection> { |
| 86 | + let mut is = CodedInputStream::from_bytes(body); |
| 87 | + let mut series = MetricSeries::new(); |
| 88 | + while let Some(tag) = is.read_raw_tag_or_eof()? { |
| 89 | + match unpack(tag)? { |
| 90 | + (1, WireType::LengthDelimited) => series.resources.push(decode_resource(&is.read_bytes()?)?), |
| 91 | + (2, WireType::LengthDelimited) => series.metric = strict(is.read_bytes()?)?, |
| 92 | + (3, WireType::LengthDelimited) => series.tags.push(to_valid_utf8(&is.read_bytes()?)), |
| 93 | + (4, WireType::LengthDelimited) => series.points.push(decode_point(&is.read_bytes()?)?), |
| 94 | + (5, WireType::Varint) => series.type_ = EnumOrUnknown::from_i32(is.read_int32()?), |
| 95 | + (6, WireType::LengthDelimited) => series.unit = strict(is.read_bytes()?)?, |
| 96 | + (7, WireType::LengthDelimited) => series.source_type_name = strict(is.read_bytes()?)?, |
| 97 | + (8, WireType::Varint) => series.interval = is.read_int64()?, |
| 98 | + (_, wire) => is.skip_field(wire)?, |
| 99 | + } |
| 100 | + } |
| 101 | + Ok(series) |
| 102 | +} |
| 103 | + |
| 104 | +fn decode_resource(body: &[u8]) -> Result<Resource, Rejection> { |
| 105 | + let mut is = CodedInputStream::from_bytes(body); |
| 106 | + let mut resource = Resource::new(); |
| 107 | + while let Some(tag) = is.read_raw_tag_or_eof()? { |
| 108 | + match unpack(tag)? { |
| 109 | + (1, WireType::LengthDelimited) => resource.type_ = strict(is.read_bytes()?)?, |
| 110 | + (2, WireType::LengthDelimited) => resource.name = strict(is.read_bytes()?)?, |
| 111 | + (_, wire) => is.skip_field(wire)?, |
| 112 | + } |
| 113 | + } |
| 114 | + Ok(resource) |
| 115 | +} |
| 116 | + |
| 117 | +fn decode_point(body: &[u8]) -> Result<MetricPoint, Rejection> { |
| 118 | + let mut is = CodedInputStream::from_bytes(body); |
| 119 | + let mut point = MetricPoint::new(); |
| 120 | + while let Some(tag) = is.read_raw_tag_or_eof()? { |
| 121 | + match unpack(tag)? { |
| 122 | + (1, WireType::Fixed64) => point.value = is.read_double()?, |
| 123 | + (2, WireType::Varint) => point.timestamp = is.read_int64()?, |
| 124 | + (_, wire) => is.skip_field(wire)?, |
| 125 | + } |
| 126 | + } |
| 127 | + Ok(point) |
| 128 | +} |
| 129 | + |
| 130 | +#[cfg(test)] |
| 131 | +mod tests { |
| 132 | + use datadog_protos::metrics::metric_payload::{MetricPoint, MetricSeries, MetricType, Resource}; |
| 133 | + use datadog_protos::metrics::MetricPayload; |
| 134 | + use protobuf::Message as _; |
| 135 | + |
| 136 | + use super::{decode_metric_payload, to_valid_utf8, Rejection}; |
| 137 | + |
| 138 | + // Non-UTF-8 in a non-tag field is the rejection production also makes, not a decode defect. |
| 139 | + #[test] |
| 140 | + fn non_utf8_name_is_a_production_faithful_rejection() { |
| 141 | + let mut payload = MetricPayload::new(); |
| 142 | + let mut series = MetricSeries::new(); |
| 143 | + series.set_metric("NAME".into()); |
| 144 | + series.set_type(MetricType::COUNT); |
| 145 | + let mut point = MetricPoint::new(); |
| 146 | + point.value = 1.0; |
| 147 | + point.timestamp = 1; |
| 148 | + series.points.push(point); |
| 149 | + payload.series.push(series); |
| 150 | + let mut bytes = payload.write_to_bytes().expect("serialize"); |
| 151 | + let pos = bytes.windows(4).position(|w| w == b"NAME").expect("marker"); |
| 152 | + for b in &mut bytes[pos..pos + 4] { |
| 153 | + *b = 0xFF; |
| 154 | + } |
| 155 | + assert_eq!(decode_metric_payload(&bytes), Err(Rejection::NonUtf8StrictField)); |
| 156 | + } |
| 157 | + |
| 158 | + // Truncated wire (field 1, LEN, incomplete length varint) is genuinely malformed. |
| 159 | + #[test] |
| 160 | + fn malformed_wire_is_rejected() { |
| 161 | + assert_eq!(decode_metric_payload(&[0x0a, 0xff]), Err(Rejection::MalformedWire)); |
| 162 | + } |
| 163 | + |
| 164 | + // Must match Go strings.ToValidUTF8: each contiguous invalid run collapses to one U+FFFD, |
| 165 | + // unlike from_utf8_lossy which emits one per maximal subpart. |
| 166 | + #[test] |
| 167 | + fn to_valid_utf8_collapses_invalid_runs() { |
| 168 | + assert_eq!(to_valid_utf8(b"ok"), "ok"); |
| 169 | + assert_eq!(to_valid_utf8(b"a\xff\xfeb"), "a\u{FFFD}b"); |
| 170 | + assert_eq!(to_valid_utf8(b"\xff\xff\xff"), "\u{FFFD}"); |
| 171 | + assert_eq!(to_valid_utf8(b"a\xffb\xffc"), "a\u{FFFD}b\u{FFFD}c"); |
| 172 | + assert_eq!(to_valid_utf8(b"caf\xc3\xa9"), "café"); |
| 173 | + } |
| 174 | + |
| 175 | + // Guards against wire-layout drift: for valid UTF-8 the lenient walker must recover |
| 176 | + // exactly what the strict parser does, or a renumbered field silently corrupts contexts. |
| 177 | + #[test] |
| 178 | + fn lenient_decode_matches_strict_for_valid_utf8() { |
| 179 | + let mut payload = MetricPayload::new(); |
| 180 | + let mut series = MetricSeries::new(); |
| 181 | + series.set_metric("adp.requests".into()); |
| 182 | + series.set_type(MetricType::COUNT); |
| 183 | + series.tags.push("env:prod".into()); |
| 184 | + series.tags.push("host:antithesis-differential".into()); |
| 185 | + series.unit = "request".into(); |
| 186 | + series.source_type_name = "dogstatsd".into(); |
| 187 | + series.interval = 10; |
| 188 | + let mut host = Resource::new(); |
| 189 | + host.set_type("host".into()); |
| 190 | + host.set_name("server-1".into()); |
| 191 | + series.resources.push(host); |
| 192 | + let mut point = MetricPoint::new(); |
| 193 | + point.value = 2.5; |
| 194 | + point.timestamp = 1_600_000_000; |
| 195 | + series.points.push(point); |
| 196 | + payload.series.push(series); |
| 197 | + |
| 198 | + let bytes = payload.write_to_bytes().expect("serialize"); |
| 199 | + let strict = MetricPayload::parse_from_bytes(&bytes).expect("strict parse"); |
| 200 | + let lenient = decode_metric_payload(&bytes).expect("lenient decode"); |
| 201 | + assert_eq!(lenient, strict); |
| 202 | + } |
| 203 | +} |
0 commit comments