netget/server/loki/
codec.rs

1//! Wire facts from the public Loki push schema and Snappy block format.
2//! No third-party protocol implementation is linked or vendored.
3use anyhow::{bail, ensure, Context, Result};
4use serde::{Deserialize, Serialize};
5use serde_json::{json, Value};
6use std::{
7    collections::BTreeMap,
8    io::{Read, Write},
9};
10pub const DEFAULT_LLM_FALLBACK: bool = false;
11pub const DEFAULT_REQUIRE_TENANT: bool = false;
12pub const MAX_BODY_BYTES: usize = 256 * 1024;
13pub const MAX_STREAMS: usize = 64;
14pub const MAX_ENTRIES: usize = 1024;
15pub const MAX_LINE_BYTES: usize = 16 * 1024;
16pub const MAX_LABELS: usize = 32;
17pub const MAX_METADATA: usize = 64;
18pub const MAX_NAME_BYTES: usize = 128;
19pub const MAX_VALUE_BYTES: usize = 2048;
20pub const MAX_PROTO_FIELDS: usize = 16384;
21#[derive(Clone, Copy, Debug, Default, Serialize, Deserialize, PartialEq, Eq)]
22#[serde(rename_all = "snake_case")]
23pub enum Encoding {
24    #[default]
25    Json,
26    GzipJson,
27    SnappyProtobuf,
28}
29impl Encoding {
30    pub fn as_str(self) -> &'static str {
31        match self {
32            Self::Json => "json",
33            Self::GzipJson => "gzip_json",
34            Self::SnappyProtobuf => "snappy_protobuf",
35        }
36    }
37}
38#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)]
39#[serde(deny_unknown_fields)]
40pub struct Entry {
41    pub timestamp_ns: i64,
42    pub line: String,
43    #[serde(default)]
44    pub structured_metadata: BTreeMap<String, String>,
45}
46#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)]
47#[serde(deny_unknown_fields)]
48pub struct Stream {
49    pub labels: BTreeMap<String, String>,
50    pub entries: Vec<Entry>,
51}
52#[derive(Clone, Debug, Serialize, Deserialize)]
53#[serde(deny_unknown_fields)]
54pub struct PushBatch {
55    #[serde(default)]
56    pub tenant_id: Option<String>,
57    #[serde(default)]
58    pub encoding: Encoding,
59    pub streams: Vec<Stream>,
60}
61pub fn validate_token(token: &str) -> Result<()> {
62    ensure!(
63        !token.is_empty()
64            && token.len() <= 1024
65            && token.bytes().all(|b| (0x21..=0x7e).contains(&b)),
66        "token printable ASCII byte limit"
67    );
68    Ok(())
69}
70pub fn validate_tenant(tenant: &str) -> Result<()> {
71    ensure!(
72        !tenant.is_empty()
73            && tenant.len() <= 150
74            && !matches!(tenant, "." | "..")
75            && tenant
76                .bytes()
77                .all(|b| b.is_ascii_alphanumeric() || b"!-_.*'()".contains(&b)),
78        "single tenant ID character/byte limit"
79    );
80    Ok(())
81}
82fn name(name: &str) -> Result<()> {
83    let mut b = name.bytes();
84    ensure!(
85        !name.is_empty()
86            && name.len() <= MAX_NAME_BYTES
87            && b.next()
88                .is_some_and(|c| c.is_ascii_alphabetic() || c == b'_')
89            && b.all(|c| c.is_ascii_alphanumeric() || c == b'_'),
90        "label/metadata name character/byte limit"
91    );
92    Ok(())
93}
94fn labels(map: &BTreeMap<String, String>, metadata: bool) -> Result<()> {
95    ensure!(
96        map.len() <= if metadata { MAX_METADATA } else { MAX_LABELS },
97        "label/metadata count limit"
98    );
99    if !metadata {
100        ensure!(!map.is_empty(), "stream requires labels");
101    }
102    for (k, v) in map {
103        name(k)?;
104        ensure!(
105            metadata || !k.starts_with("__"),
106            "reserved stream label namespace"
107        );
108        ensure!(
109            v.len() <= MAX_VALUE_BYTES,
110            "label/metadata value byte limit"
111        );
112    }
113    Ok(())
114}
115pub fn validate_batch(batch: &PushBatch) -> Result<()> {
116    if let Some(tenant) = &batch.tenant_id {
117        validate_tenant(tenant)?;
118    }
119    ensure!(
120        !batch.streams.is_empty() && batch.streams.len() <= MAX_STREAMS,
121        "stream count limit"
122    );
123    let mut total = 0;
124    let mut proto_fields = 0usize;
125    let mut seen = std::collections::BTreeSet::new();
126    for stream in &batch.streams {
127        labels(&stream.labels, false)?;
128        proto_fields += 2;
129        ensure!(seen.insert(&stream.labels), "duplicate stream label set");
130        ensure!(!stream.entries.is_empty(), "stream requires entries");
131        total += stream.entries.len();
132        ensure!(total <= MAX_ENTRIES, "entry count limit");
133        for e in &stream.entries {
134            ensure!(e.line.len() <= MAX_LINE_BYTES, "log line byte limit");
135            labels(&e.structured_metadata, true)?;
136            proto_fields += 5 + 3 * e.structured_metadata.len();
137            ensure!(
138                batch.encoding != Encoding::SnappyProtobuf || proto_fields <= MAX_PROTO_FIELDS,
139                "protobuf field count limit"
140            );
141        }
142    }
143    Ok(())
144}
145pub fn encode_batch(batch: &PushBatch) -> Result<Vec<u8>> {
146    validate_batch(batch)?;
147    let bytes = match batch.encoding {
148        Encoding::SnappyProtobuf => encode_proto(&batch.streams),
149        _ => serde_json::to_vec(
150            &json!({"streams":batch.streams.iter().map(|s|json!({"stream":s.labels,"values":s.entries.iter().map(|e|if e.structured_metadata.is_empty(){json!([e.timestamp_ns.to_string(),e.line])}else{json!([e.timestamp_ns.to_string(),e.line,e.structured_metadata])}).collect::<Vec<_>>()})).collect::<Vec<_>>()}),
151        )?,
152    };
153    ensure!(
154        bytes.len() <= MAX_BODY_BYTES,
155        "uncompressed body byte limit"
156    );
157    let bytes = match batch.encoding {
158        Encoding::Json => bytes,
159        Encoding::SnappyProtobuf => snappy_encode(&bytes),
160        Encoding::GzipJson => {
161            let mut gzip =
162                flate2::write::GzEncoder::new(Vec::new(), flate2::Compression::default());
163            gzip.write_all(&bytes)?;
164            gzip.finish()?
165        }
166    };
167    ensure!(bytes.len() <= MAX_BODY_BYTES, "wire body byte limit");
168    Ok(bytes)
169}
170pub fn decode_batch(bytes: &[u8], encoding: Encoding) -> Result<Vec<Stream>> {
171    ensure!(bytes.len() <= MAX_BODY_BYTES, "wire body byte limit");
172    let decoded = match encoding {
173        Encoding::Json => bytes.to_vec(),
174        Encoding::SnappyProtobuf => snappy_decode(bytes)?,
175        Encoding::GzipJson => {
176            let mut decoder = flate2::read::MultiGzDecoder::new(bytes);
177            let mut out = Vec::new();
178            decoder
179                .by_ref()
180                .take((MAX_BODY_BYTES + 1) as u64)
181                .read_to_end(&mut out)
182                .context("invalid gzip")?;
183            ensure!(out.len() <= MAX_BODY_BYTES, "gzip decompressed byte limit");
184            out
185        }
186    };
187    let streams = match encoding {
188        Encoding::SnappyProtobuf => decode_proto(&decoded)?,
189        _ => decode_json(&decoded)?,
190    };
191    validate_batch(&PushBatch {
192        tenant_id: None,
193        encoding,
194        streams: streams.clone(),
195    })?;
196    Ok(streams)
197}
198// Reject duplicate JSON keys before converting the public wire shape. Parsing depth is
199// separately bounded so an ignored/unknown field cannot smuggle a nesting bomb.
200struct StrictValue(Value);
201impl<'de> Deserialize<'de> for StrictValue {
202    fn deserialize<D: serde::Deserializer<'de>>(d: D) -> std::result::Result<Self, D::Error> {
203        use serde::de::{MapAccess, SeqAccess, Visitor};
204        struct V;
205        impl<'de> Visitor<'de> for V {
206            type Value = StrictValue;
207            fn expecting(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
208                f.write_str("JSON value with unique object keys")
209            }
210            fn visit_bool<E: serde::de::Error>(
211                self,
212                v: bool,
213            ) -> std::result::Result<Self::Value, E> {
214                Ok(StrictValue(json!(v)))
215            }
216            fn visit_i64<E: serde::de::Error>(self, v: i64) -> std::result::Result<Self::Value, E> {
217                Ok(StrictValue(json!(v)))
218            }
219            fn visit_u64<E: serde::de::Error>(self, v: u64) -> std::result::Result<Self::Value, E> {
220                Ok(StrictValue(json!(v)))
221            }
222            fn visit_f64<E: serde::de::Error>(self, v: f64) -> std::result::Result<Self::Value, E> {
223                Ok(StrictValue(json!(v)))
224            }
225            fn visit_str<E: serde::de::Error>(
226                self,
227                v: &str,
228            ) -> std::result::Result<Self::Value, E> {
229                Ok(StrictValue(json!(v)))
230            }
231            fn visit_string<E: serde::de::Error>(
232                self,
233                v: String,
234            ) -> std::result::Result<Self::Value, E> {
235                Ok(StrictValue(json!(v)))
236            }
237            fn visit_unit<E: serde::de::Error>(self) -> std::result::Result<Self::Value, E> {
238                Ok(StrictValue(Value::Null))
239            }
240            fn visit_none<E: serde::de::Error>(self) -> std::result::Result<Self::Value, E> {
241                Ok(StrictValue(Value::Null))
242            }
243            fn visit_seq<A: SeqAccess<'de>>(
244                self,
245                mut a: A,
246            ) -> std::result::Result<Self::Value, A::Error> {
247                let mut out = Vec::new();
248                while let Some(v) = a.next_element::<StrictValue>()? {
249                    out.push(v.0);
250                }
251                Ok(StrictValue(Value::Array(out)))
252            }
253            fn visit_map<A: MapAccess<'de>>(
254                self,
255                mut a: A,
256            ) -> std::result::Result<Self::Value, A::Error> {
257                let mut out = serde_json::Map::new();
258                while let Some(k) = a.next_key::<String>()? {
259                    if out.contains_key(&k) {
260                        return Err(serde::de::Error::custom("duplicate JSON key"));
261                    }
262                    out.insert(k, a.next_value::<StrictValue>()?.0);
263                }
264                Ok(StrictValue(Value::Object(out)))
265            }
266        }
267        d.deserialize_any(V)
268    }
269}
270fn json_depth(bytes: &[u8]) -> Result<()> {
271    let (mut depth, mut quoted, mut escaped) = (0usize, false, false);
272    for &b in bytes {
273        if quoted {
274            if escaped {
275                escaped = false;
276            } else if b == b'\\' {
277                escaped = true;
278            } else if b == b'"' {
279                quoted = false;
280            }
281        } else {
282            match b {
283                b'"' => quoted = true,
284                b'[' | b'{' => {
285                    depth += 1;
286                    ensure!(depth <= 8, "JSON nesting depth limit");
287                }
288                b']' | b'}' => {
289                    depth = depth.saturating_sub(1);
290                }
291                _ => {}
292            }
293        }
294    }
295    Ok(())
296}
297fn decode_json(bytes: &[u8]) -> Result<Vec<Stream>> {
298    json_depth(bytes)?;
299    let value = serde_json::from_slice::<StrictValue>(bytes)?.0;
300    let mut root = value
301        .as_object()
302        .context("push JSON object required")?
303        .clone();
304    ensure!(root.len() == 1, "only streams push field supported");
305    let streams = root.remove("streams").context("streams required")?;
306    let wire = streams.as_array().context("streams array required")?;
307    ensure!(wire.len() <= MAX_STREAMS, "stream count limit");
308    let mut out = Vec::new();
309    let mut count = 0;
310    for stream in wire {
311        let o = stream.as_object().context("stream object required")?;
312        ensure!(o.len() == 2, "stream and values required only");
313        let map: BTreeMap<String, String> =
314            serde_json::from_value(o.get("stream").context("stream labels required")?.clone())?;
315        let values = o
316            .get("values")
317            .and_then(Value::as_array)
318            .context("values array required")?;
319        count += values.len();
320        ensure!(count <= MAX_ENTRIES, "entry count limit");
321        let mut entries = Vec::new();
322        for value in values {
323            let arr = value.as_array().context("log value array required")?;
324            ensure!(
325                matches!(arr.len(), 2 | 3),
326                "log value requires timestamp string,line and optional metadata"
327            );
328            let text = arr[0]
329                .as_str()
330                .context("timestamp must be decimal nanoseconds STRING")?;
331            ensure!(
332                !text.is_empty()
333                    && text.len() <= 20
334                    && text
335                        .bytes()
336                        .enumerate()
337                        .all(|(i, b)| b.is_ascii_digit() || (i == 0 && b == b'-')),
338                "decimal nanoseconds timestamp required"
339            );
340            let timestamp_ns = text
341                .parse::<i64>()
342                .context("timestamp signed nanoseconds range")?;
343            let line = arr[1].as_str().context("line must be string")?.to_owned();
344            let structured_metadata = if arr.len() == 3 {
345                serde_json::from_value(arr[2].clone())?
346            } else {
347                BTreeMap::new()
348            };
349            entries.push(Entry {
350                timestamp_ns,
351                line,
352                structured_metadata,
353            });
354        }
355        out.push(Stream {
356            labels: map,
357            entries,
358        });
359    }
360    Ok(out)
361}
362fn varint(out: &mut Vec<u8>, mut n: u64) {
363    while n >= 128 {
364        out.push((n as u8) | 128);
365        n >>= 7;
366    }
367    out.push(n as u8);
368}
369fn message(out: &mut Vec<u8>, field: u64, bytes: &[u8]) {
370    varint(out, (field << 3) | 2);
371    varint(out, bytes.len() as u64);
372    out.extend_from_slice(bytes);
373}
374fn integer(out: &mut Vec<u8>, field: u64, n: u64) {
375    varint(out, field << 3);
376    varint(out, n);
377}
378fn label_text(labels: &BTreeMap<String, String>) -> String {
379    format!(
380        "{{{}}}",
381        labels
382            .iter()
383            .map(|(k, v)| format!("{k}={}", serde_json::to_string(v).unwrap()))
384            .collect::<Vec<_>>()
385            .join(",")
386    )
387}
388fn encode_proto(streams: &[Stream]) -> Vec<u8> {
389    let mut out = Vec::new();
390    for s in streams {
391        let mut wire = Vec::new();
392        message(&mut wire, 1, label_text(&s.labels).as_bytes());
393        for e in &s.entries {
394            let mut entry = Vec::new();
395            let mut timestamp = Vec::new();
396            integer(
397                &mut timestamp,
398                1,
399                e.timestamp_ns.div_euclid(1_000_000_000) as u64,
400            );
401            integer(
402                &mut timestamp,
403                2,
404                e.timestamp_ns.rem_euclid(1_000_000_000) as u64,
405            );
406            message(&mut entry, 1, &timestamp);
407            message(&mut entry, 2, e.line.as_bytes());
408            for (k, v) in &e.structured_metadata {
409                let mut pair = Vec::new();
410                message(&mut pair, 1, k.as_bytes());
411                message(&mut pair, 2, v.as_bytes());
412                message(&mut entry, 3, &pair);
413            }
414            message(&mut wire, 2, &entry);
415        }
416        message(&mut out, 1, &wire);
417    }
418    out
419}
420struct Reader<'a> {
421    bytes: &'a [u8],
422    at: usize,
423}
424impl<'a> Reader<'a> {
425    fn new(bytes: &'a [u8]) -> Self {
426        Self { bytes, at: 0 }
427    }
428    fn take(&mut self, n: usize) -> Result<&'a [u8]> {
429        let end = self.at.checked_add(n).context("length overflow")?;
430        let v = self.bytes.get(self.at..end).context("truncated field")?;
431        self.at = end;
432        Ok(v)
433    }
434    fn number(&mut self) -> Result<u64> {
435        let mut n = 0;
436        for shift in (0..70).step_by(7) {
437            let b = self.take(1)?[0];
438            ensure!(shift != 63 || b <= 1, "varint overflow");
439            n |= u64::from(b & 127) << shift;
440            if b & 128 == 0 {
441                return Ok(n);
442            }
443        }
444        bail!("varint overflow")
445    }
446    fn field(&mut self, budget: &mut usize) -> Result<Option<(u64, u8)>> {
447        if self.at == self.bytes.len() {
448            return Ok(None);
449        }
450        *budget = budget
451            .checked_sub(1)
452            .context("protobuf field count limit")?;
453        let key = self.number()?;
454        ensure!(
455            key >> 3 > 0 && key >> 3 <= 0x1fffffff,
456            "invalid protobuf field number"
457        );
458        Ok(Some((key >> 3, (key & 7) as u8)))
459    }
460    fn blob(&mut self, wire: u8) -> Result<&'a [u8]> {
461        ensure!(wire == 2, "expected length-delimited field");
462        let n = usize::try_from(self.number()?)?;
463        self.take(n)
464    }
465    fn scalar(&mut self, wire: u8) -> Result<u64> {
466        ensure!(wire == 0, "expected integer field");
467        self.number()
468    }
469    fn skip(&mut self, wire: u8) -> Result<()> {
470        match wire {
471            0 => {
472                self.number()?;
473            }
474            1 => {
475                self.take(8)?;
476            }
477            2 => {
478                self.blob(wire)?;
479            }
480            5 => {
481                self.take(4)?;
482            }
483            _ => bail!("protobuf groups/invalid wire types unsupported"),
484        };
485        Ok(())
486    }
487}
488fn text(bytes: &[u8]) -> Result<String> {
489    Ok(std::str::from_utf8(bytes)?.to_owned())
490}
491fn timestamp(bytes: &[u8], budget: &mut usize) -> Result<i64> {
492    let mut r = Reader::new(bytes);
493    let (mut seconds, mut nanos) = (None, None);
494    while let Some((field, wire)) = r.field(budget)? {
495        match field {
496            1 => {
497                ensure!(seconds.is_none(), "duplicate timestamp seconds");
498                seconds = Some(r.scalar(wire)? as i64);
499            }
500            2 => {
501                ensure!(nanos.is_none(), "duplicate timestamp nanos");
502                nanos = Some(r.scalar(wire)?);
503            }
504            _ => r.skip(wire)?,
505        }
506    }
507    let nanos = nanos.unwrap_or(0);
508    ensure!(nanos < 1_000_000_000, "timestamp nanos range");
509    Ok(i64::try_from(
510        i128::from(seconds.unwrap_or(0)) * 1_000_000_000 + i128::from(nanos),
511    )?)
512}
513fn pair(bytes: &[u8], budget: &mut usize) -> Result<(String, String)> {
514    let mut r = Reader::new(bytes);
515    let (mut name, mut value) = (None, None);
516    while let Some((field, wire)) = r.field(budget)? {
517        match field {
518            1 => {
519                ensure!(name.is_none(), "duplicate metadata name");
520                name = Some(text(r.blob(wire)?)?);
521            }
522            2 => {
523                ensure!(value.is_none(), "duplicate metadata value");
524                value = Some(text(r.blob(wire)?)?);
525            }
526            _ => r.skip(wire)?,
527        }
528    }
529    Ok((
530        name.context("metadata name required")?,
531        value.unwrap_or_default(),
532    ))
533}
534fn entry(bytes: &[u8], budget: &mut usize) -> Result<Entry> {
535    let mut r = Reader::new(bytes);
536    let (mut ts, mut line) = (None, None);
537    let mut metadata = BTreeMap::new();
538    while let Some((field, wire)) = r.field(budget)? {
539        match field {
540            1 => {
541                ensure!(ts.is_none(), "duplicate entry timestamp");
542                ts = Some(timestamp(r.blob(wire)?, budget)?);
543            }
544            2 => {
545                ensure!(line.is_none(), "duplicate entry line");
546                line = Some(text(r.blob(wire)?)?);
547            }
548            3 => {
549                let (k, v) = pair(r.blob(wire)?, budget)?;
550                ensure!(metadata.insert(k, v).is_none(), "duplicate metadata name");
551                ensure!(metadata.len() <= MAX_METADATA, "metadata count limit");
552            }
553            4 => bail!("parsed query metadata is not a push field"),
554            _ => r.skip(wire)?,
555        }
556    }
557    Ok(Entry {
558        timestamp_ns: ts.context("entry timestamp required")?,
559        line: line.unwrap_or_default(),
560        structured_metadata: metadata,
561    })
562}
563fn decode_proto(bytes: &[u8]) -> Result<Vec<Stream>> {
564    let mut budget = MAX_PROTO_FIELDS;
565    let mut r = Reader::new(bytes);
566    let mut out = Vec::new();
567    let mut total = 0;
568    while let Some((field, wire)) = r.field(&mut budget)? {
569        match field {
570            1 => {
571                ensure!(out.len() < MAX_STREAMS, "stream count limit");
572                let mut s = Reader::new(r.blob(wire)?);
573                let mut labels = None;
574                let mut entries = Vec::new();
575                while let Some((field, wire)) = s.field(&mut budget)? {
576                    match field {
577                        1 => {
578                            ensure!(labels.is_none(), "duplicate stream labels");
579                            labels = Some(parse_labels(&text(s.blob(wire)?)?)?);
580                        }
581                        2 => {
582                            total += 1;
583                            ensure!(total <= MAX_ENTRIES, "entry count limit");
584                            entries.push(entry(s.blob(wire)?, &mut budget)?);
585                        }
586                        3 => {
587                            s.scalar(wire)?;
588                        }
589                        _ => s.skip(wire)?,
590                    }
591                }
592                out.push(Stream {
593                    labels: labels.context("stream labels required")?,
594                    entries,
595                });
596            }
597            2 => ensure!(
598                matches!(text(r.blob(wire)?)?.as_str(), "" | "loki"),
599                "only Loki push format supported"
600            ),
601            _ => r.skip(wire)?,
602        }
603    }
604    Ok(out)
605}
606// Prometheus label-set text uses Go quoted strings, including octal/hex escapes.
607fn parse_labels(s: &str) -> Result<BTreeMap<String, String>> {
608    let b = s.as_bytes();
609    let mut at = 0;
610    let mut out = BTreeMap::new();
611    fn ws(b: &[u8], at: &mut usize) {
612        while b.get(*at).is_some_and(u8::is_ascii_whitespace) {
613            *at += 1;
614        }
615    }
616    ws(b, &mut at);
617    ensure!(b.get(at) == Some(&b'{'), "label set requires braces");
618    at += 1;
619    loop {
620        ws(b, &mut at);
621        if b.get(at) == Some(&b'}') {
622            at += 1;
623            break;
624        }
625        let start = at;
626        while b
627            .get(at)
628            .is_some_and(|c| c.is_ascii_alphanumeric() || *c == b'_')
629        {
630            at += 1;
631        }
632        let key = std::str::from_utf8(&b[start..at])?.to_owned();
633        name(&key)?;
634        ws(b, &mut at);
635        ensure!(b.get(at) == Some(&b'='), "label equality required");
636        at += 1;
637        ws(b, &mut at);
638        let value = go_string(b, &mut at)?;
639        ensure!(out.insert(key, value).is_none(), "duplicate stream label");
640        ensure!(out.len() <= MAX_LABELS, "label count limit");
641        ws(b, &mut at);
642        match b.get(at) {
643            Some(b',') => at += 1,
644            Some(b'}') => {
645                at += 1;
646                break;
647            }
648            _ => bail!("label comma/end required"),
649        }
650    }
651    ws(b, &mut at);
652    ensure!(at == b.len(), "trailing label text");
653    Ok(out)
654}
655fn go_string(bytes: &[u8], at: &mut usize) -> Result<String> {
656    ensure!(bytes.get(*at) == Some(&b'"'), "quoted label value required");
657    *at += 1;
658    let mut out = Vec::new();
659    loop {
660        let c = *bytes.get(*at).context("unterminated label value")?;
661        *at += 1;
662        match c {
663            b'"' => break,
664            b'\\' => {
665                let e = *bytes.get(*at).context("truncated label escape")?;
666                *at += 1;
667                match e {
668                    b'a' => out.push(7),
669                    b'b' => out.push(8),
670                    b'f' => out.push(12),
671                    b'n' => out.push(10),
672                    b'r' => out.push(13),
673                    b't' => out.push(9),
674                    b'v' => out.push(11),
675                    b'\\' | b'"' => out.push(e),
676                    b'x' | b'u' | b'U' => {
677                        let n = match e {
678                            b'x' => 2,
679                            b'u' => 4,
680                            _ => 8,
681                        };
682                        let end = at.checked_add(n).context("escape length")?;
683                        let raw = std::str::from_utf8(
684                            bytes.get(*at..end).context("truncated hex escape")?,
685                        )?;
686                        let value = u32::from_str_radix(raw, 16)?;
687                        *at = end;
688                        if e == b'x' {
689                            out.push(u8::try_from(value)?);
690                        } else {
691                            let ch = char::from_u32(value).context("invalid Unicode scalar")?;
692                            let mut buf = [0; 4];
693                            out.extend_from_slice(ch.encode_utf8(&mut buf).as_bytes());
694                        }
695                    }
696                    b'0'..=b'7' => {
697                        let end = at.checked_add(2).context("escape length")?;
698                        let rest = bytes.get(*at..end).context("truncated octal escape")?;
699                        ensure!(
700                            rest.iter().all(|b| (b'0'..=b'7').contains(b)),
701                            "invalid octal escape"
702                        );
703                        let n = u32::from(e - b'0') * 64
704                            + u32::from(rest[0] - b'0') * 8
705                            + u32::from(rest[1] - b'0');
706                        out.push(u8::try_from(n)?);
707                        *at = end;
708                    }
709                    _ => bail!("unknown label escape"),
710                }
711            }
712            b'\n' | b'\r' => bail!("unescaped newline in label value"),
713            _ => out.push(c),
714        }
715    }
716    Ok(String::from_utf8(out)?)
717}
718pub fn snappy_encode(bytes: &[u8]) -> Vec<u8> {
719    let mut out = Vec::new();
720    varint(&mut out, bytes.len() as u64);
721    if !bytes.is_empty() {
722        let n = bytes.len() - 1;
723        if n < 60 {
724            out.push((n as u8) << 2);
725        } else {
726            let width = ((usize::BITS - n.leading_zeros()) as usize).div_ceil(8);
727            out.push(((59 + width) as u8) << 2);
728            out.extend_from_slice(&(n as u32).to_le_bytes()[..width]);
729        }
730        out.extend_from_slice(bytes);
731    }
732    out
733}
734pub fn snappy_decode(bytes: &[u8]) -> Result<Vec<u8>> {
735    let mut r = Reader::new(bytes);
736    let n = usize::try_from(r.number()?)?;
737    ensure!(r.at <= 5, "Snappy length preamble overflow");
738    ensure!(n <= MAX_BODY_BYTES, "Snappy decompressed byte limit");
739    let mut out = Vec::with_capacity(n);
740    while r.at < bytes.len() {
741        let tag = r.take(1)?[0];
742        let (kind, len, offset) = match tag & 3 {
743            0 => {
744                let v = usize::from(tag >> 2);
745                let len = if v < 60 {
746                    v + 1
747                } else {
748                    let width = v - 59;
749                    let mut raw = [0; 4];
750                    raw[..width].copy_from_slice(r.take(width)?);
751                    usize::try_from(u32::from_le_bytes(raw))?
752                        .checked_add(1)
753                        .context("literal length overflow")?
754                };
755                (0, len, 0)
756            }
757            1 => (
758                1,
759                usize::from((tag >> 2) & 7) + 4,
760                (usize::from(tag & 0xe0) << 3) | usize::from(r.take(1)?[0]),
761            ),
762            2 => {
763                let b = r.take(2)?;
764                (
765                    2,
766                    usize::from(tag >> 2) + 1,
767                    usize::from(u16::from_le_bytes([b[0], b[1]])),
768                )
769            }
770            _ => {
771                let b = r.take(4)?;
772                (
773                    3,
774                    usize::from(tag >> 2) + 1,
775                    usize::try_from(u32::from_le_bytes([b[0], b[1], b[2], b[3]]))?,
776                )
777            }
778        };
779        ensure!(
780            len <= n.saturating_sub(out.len()),
781            "Snappy output length mismatch/limit"
782        );
783        if kind == 0 {
784            out.extend_from_slice(r.take(len)?);
785        } else {
786            ensure!(
787                offset > 0 && offset <= out.len(),
788                "invalid Snappy backreference"
789            );
790            for _ in 0..len {
791                out.push(out[out.len() - offset]);
792            }
793        }
794    }
795    ensure!(out.len() == n, "Snappy truncated output");
796    Ok(out)
797}
netget/server/loki/codec.rs

netget/server/loki/
codec.rs
