netget/server/gelf/
codec.rs

1//! Bounded GELF 1.1 JSON, compression, TCP framing and source-correlated UDP chunks.
2use anyhow::{bail, ensure, Context, Result};
3use flate2::{
4    bufread::GzDecoder,
5    write::{GzEncoder, ZlibEncoder},
6    Compression as FlateCompression,
7};
8use serde::{Deserialize, Serialize};
9use serde_json::{Map, Value};
10use std::{
11    collections::{BTreeMap, HashMap},
12    io::{Read, Write},
13    net::SocketAddr,
14    time::Duration,
15};
16use tokio::time::Instant;
17
18pub const DEFAULT_TRANSPORT: &str = "udp";
19pub const DEFAULT_COMPRESSION: &str = "auto";
20pub const DEFAULT_CHUNK_SIZE: usize = 1420;
21pub const DEFAULT_LLM_FALLBACK: bool = false;
22pub const MAX_MESSAGE_BYTES: usize = 256 * 1024;
23pub const MAX_DATAGRAM_BYTES: usize = 8192;
24pub const MAX_CHUNKS: usize = 128;
25pub const MAX_PENDING_MESSAGES: usize = 128;
26pub const MAX_PENDING_BYTES: usize = 4 * 1024 * 1024;
27pub const MAX_RECENT_IDS: usize = 256;
28pub const CHUNK_TIMEOUT: Duration = Duration::from_secs(5);
29
30#[derive(Clone, Copy, Debug, PartialEq, Eq)]
31pub enum Transport {
32    Udp,
33    Tcp,
34}
35impl Transport {
36    pub fn parse(s: &str) -> Result<Self> {
37        match s {
38            "udp" => Ok(Self::Udp),
39            "tcp" => Ok(Self::Tcp),
40            _ => bail!("transport must be udp or tcp"),
41        }
42    }
43    pub fn as_str(self) -> &'static str {
44        match self {
45            Self::Udp => "udp",
46            Self::Tcp => "tcp",
47        }
48    }
49}
50#[derive(Clone, Copy, Debug)]
51pub enum Compression {
52    None,
53    Gzip,
54    Zlib,
55}
56impl Compression {
57    pub fn parse(s: &str, transport: Transport) -> Result<Self> {
58        match (s, transport) {
59            ("auto", Transport::Udp) | ("gzip", Transport::Udp) => Ok(Self::Gzip),
60            ("auto" | "none", Transport::Tcp) | ("none", Transport::Udp) => Ok(Self::None),
61            ("zlib", Transport::Udp) => Ok(Self::Zlib),
62            ("gzip" | "zlib", Transport::Tcp) => bail!("GELF TCP does not permit compression"),
63            _ => bail!("compression must be auto, none, gzip or zlib"),
64        }
65    }
66}
67#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
68#[serde(deny_unknown_fields)]
69pub struct Message {
70    pub host: String,
71    pub short_message: String,
72    #[serde(default, skip_serializing_if = "Option::is_none")]
73    pub full_message: Option<String>,
74    #[serde(default, skip_serializing_if = "Option::is_none")]
75    pub timestamp: Option<f64>,
76    #[serde(default, skip_serializing_if = "Option::is_none")]
77    pub level: Option<u8>,
78    #[serde(default, skip_serializing_if = "Option::is_none")]
79    pub facility: Option<String>,
80    #[serde(default, skip_serializing_if = "Option::is_none")]
81    pub file: Option<String>,
82    #[serde(default, skip_serializing_if = "Option::is_none")]
83    pub line: Option<u64>,
84    /// Names have no leading underscore; the wire encoder adds it.
85    #[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
86    pub additional_fields: BTreeMap<String, Value>,
87}
88fn validate(m: &Message) -> Result<()> {
89    ensure!(
90        !m.host.is_empty() && !m.short_message.is_empty(),
91        "host and short_message must be nonempty"
92    );
93    if let Some(ts) = m.timestamp {
94        ensure!(
95            ts.is_finite() && ts >= 0.0,
96            "timestamp must be finite nonnegative UNIX seconds"
97        );
98    }
99    if let Some(level) = m.level {
100        ensure!(level <= 7, "level must be 0..7");
101    }
102    for (key, value) in &m.additional_fields {
103        ensure!(
104            !key.is_empty()
105                && key != "id"
106                && key
107                    .chars()
108                    .all(|c| c.is_alphanumeric() || matches!(c, '_' | '.' | '-')),
109            "invalid additional field name"
110        );
111        ensure!(
112            value.is_string() || value.is_number(),
113            "additional fields must be strings or numbers"
114        );
115    }
116    Ok(())
117}
118pub fn encode_json(m: &Message) -> Result<Vec<u8>> {
119    validate(m)?;
120    let mut object = serde_json::to_value(m)?
121        .as_object()
122        .context("message object")?
123        .clone();
124    object.remove("additional_fields");
125    object.insert("version".into(), Value::String("1.1".into()));
126    for (key, value) in &m.additional_fields {
127        object.insert(format!("_{key}"), value.clone());
128    }
129    let bytes = serde_json::to_vec(&object)?;
130    ensure!(
131        bytes.len() <= MAX_MESSAGE_BYTES,
132        "GELF message exceeds {MAX_MESSAGE_BYTES} bytes"
133    );
134    Ok(bytes)
135}
136pub fn parse_json(bytes: &[u8]) -> Result<Message> {
137    ensure!(
138        !bytes.is_empty() && bytes.len() <= MAX_MESSAGE_BYTES,
139        "GELF message byte limit"
140    );
141    let mut object: Map<String, Value> =
142        serde_json::from_slice(bytes).context("invalid GELF JSON object")?;
143    ensure!(
144        object.remove("version") == Some(Value::String("1.1".into())),
145        "GELF version must be 1.1"
146    );
147    let mut extras = BTreeMap::new();
148    let keys: Vec<_> = object
149        .keys()
150        .filter(|k| k.starts_with('_'))
151        .cloned()
152        .collect();
153    for key in keys {
154        extras.insert(key[1..].to_owned(), object.remove(&key).unwrap());
155    }
156    ensure!(
157        !object.contains_key("additional_fields"),
158        "unprefixed additional_fields is not a GELF wire field"
159    );
160    let mut m: Message =
161        serde_json::from_value(Value::Object(object)).context("invalid GELF fields")?;
162    m.additional_fields = extras;
163    validate(&m)?;
164    Ok(m)
165}
166pub fn decompress(bytes: &[u8]) -> Result<Message> {
167    ensure!(
168        !bytes.is_empty() && bytes.len() <= MAX_MESSAGE_BYTES,
169        "compressed message byte limit"
170    );
171    let decoded = if bytes.starts_with(&[0x1f, 0x8b]) {
172        let mut decoder = GzDecoder::new(bytes);
173        let mut out = Vec::new();
174        decoder
175            .by_ref()
176            .take((MAX_MESSAGE_BYTES + 1) as u64)
177            .read_to_end(&mut out)?;
178        ensure!(
179            out.len() <= MAX_MESSAGE_BYTES,
180            "decompressed message byte limit"
181        );
182        ensure!(
183            decoder.get_ref().is_empty(),
184            "trailing bytes after gzip message"
185        );
186        out
187    } else if bytes.len() >= 2
188        && bytes[0] & 0x0f == 8
189        && (u16::from_be_bytes([bytes[0], bytes[1]]) % 31 == 0)
190    {
191        // The streaming Read adapter can report EOF before the zlib checksum
192        // arrives. Require an actual StreamEnd, complete trailer and exact input.
193        let mut decoder = flate2::Decompress::new(true);
194        let mut out = vec![0u8; MAX_MESSAGE_BYTES + 1];
195        let status = decoder.decompress(bytes, &mut out, flate2::FlushDecompress::Finish)?;
196        ensure!(
197            status == flate2::Status::StreamEnd,
198            "incomplete or oversized zlib message"
199        );
200        ensure!(
201            decoder.total_in() as usize == bytes.len(),
202            "trailing bytes after zlib message"
203        );
204        ensure!(
205            decoder.total_out() as usize <= MAX_MESSAGE_BYTES,
206            "decompressed message byte limit"
207        );
208        out.truncate(decoder.total_out() as usize);
209        out
210    } else {
211        bytes.to_vec()
212    };
213    parse_json(&decoded)
214}
215pub fn encode_udp(
216    m: &Message,
217    compression: Compression,
218    chunk_size: usize,
219    id: [u8; 8],
220) -> Result<Vec<Vec<u8>>> {
221    ensure!(
222        (13..=MAX_DATAGRAM_BYTES).contains(&chunk_size),
223        "chunk_size must be 13..={MAX_DATAGRAM_BYTES}"
224    );
225    let json = encode_json(m)?;
226    let bytes = match compression {
227        Compression::None => json,
228        Compression::Gzip => {
229            let mut encoder = GzEncoder::new(Vec::new(), FlateCompression::default());
230            encoder.write_all(&json)?;
231            encoder.finish()?
232        }
233        Compression::Zlib => {
234            let mut encoder = ZlibEncoder::new(Vec::new(), FlateCompression::default());
235            encoder.write_all(&json)?;
236            encoder.finish()?
237        }
238    };
239    ensure!(
240        bytes.len() <= MAX_MESSAGE_BYTES,
241        "compressed message byte limit"
242    );
243    if bytes.len() <= chunk_size {
244        return Ok(vec![bytes]);
245    }
246    let payload_size = chunk_size - 12;
247    let count = bytes.len().div_ceil(payload_size);
248    ensure!(
249        count <= MAX_CHUNKS,
250        "message requires more than {MAX_CHUNKS} chunks"
251    );
252    Ok(bytes
253        .chunks(payload_size)
254        .enumerate()
255        .map(|(seq, payload)| {
256            let mut out = Vec::with_capacity(12 + payload.len());
257            out.extend_from_slice(&[0x1e, 0x0f]);
258            out.extend_from_slice(&id);
259            out.push(seq as u8);
260            out.push(count as u8);
261            out.extend_from_slice(payload);
262            out
263        })
264        .collect())
265}
266struct Pending {
267    started: Instant,
268    count: usize,
269    bytes: usize,
270    parts: Vec<Option<Vec<u8>>>,
271}
272type Key = (SocketAddr, [u8; 8]);
273#[derive(Default)]
274pub struct Reassembler {
275    pending: HashMap<Key, Pending>,
276    recent: HashMap<Key, Instant>,
277    bytes: usize,
278}
279impl Reassembler {
280    pub fn recent_count(&self) -> usize {
281        self.recent.len()
282    }
283    pub fn pending_count(&self) -> usize {
284        self.pending.len()
285    }
286    pub fn pending_bytes(&self) -> usize {
287        self.bytes
288    }
289    pub fn expire(&mut self, now: Instant) {
290        self.recent.retain(|_, until| *until > now);
291        let expired: Vec<_> = self
292            .pending
293            .iter()
294            .filter(|(_, p)| now.duration_since(p.started) >= CHUNK_TIMEOUT)
295            .map(|(k, _)| *k)
296            .collect();
297        for key in expired {
298            self.drop_pending(key);
299            self.remember(key, now);
300        }
301    }
302    fn drop_pending(&mut self, key: Key) {
303        if let Some(p) = self.pending.remove(&key) {
304            self.bytes -= p.bytes;
305        }
306    }
307    fn remember(&mut self, key: Key, now: Instant) {
308        if self.recent.len() == MAX_RECENT_IDS {
309            if let Some(old) = self.recent.iter().min_by_key(|(_, v)| **v).map(|(k, _)| *k) {
310                self.recent.remove(&old);
311            }
312        }
313        self.recent.insert(key, now + CHUNK_TIMEOUT);
314    }
315    pub fn push(
316        &mut self,
317        peer: SocketAddr,
318        bytes: &[u8],
319        now: Instant,
320    ) -> Result<Option<Message>> {
321        self.expire(now);
322        ensure!(
323            !bytes.is_empty() && bytes.len() <= MAX_DATAGRAM_BYTES,
324            "GELF datagram byte limit"
325        );
326        if !bytes.starts_with(&[0x1e, 0x0f]) {
327            return decompress(bytes).map(Some);
328        }
329        ensure!(bytes.len() > 12, "empty or truncated chunk");
330        let id: [u8; 8] = bytes[2..10].try_into().unwrap();
331        let key = (peer, id);
332        let seq = bytes[10] as usize;
333        let count = bytes[11] as usize;
334        ensure!(
335            (1..=MAX_CHUNKS).contains(&count) && seq < count,
336            "invalid chunk sequence or count"
337        );
338        if self.recent.contains_key(&key) {
339            return Ok(None);
340        }
341        if !self.pending.contains_key(&key) {
342            ensure!(
343                self.pending.len() < MAX_PENDING_MESSAGES,
344                "pending message count limit"
345            );
346            self.pending.insert(
347                key,
348                Pending {
349                    started: now,
350                    count,
351                    bytes: 0,
352                    parts: vec![None; count],
353                },
354            );
355        }
356        let pending = self.pending.get(&key).unwrap();
357        if pending.count != count
358            || pending.parts[seq]
359                .as_ref()
360                .is_some_and(|p| p != &bytes[12..])
361        {
362            self.drop_pending(key);
363            self.remember(key, now);
364            bail!("conflicting GELF chunks");
365        }
366        if pending.parts[seq].is_some() {
367            return Ok(None);
368        }
369        if pending.bytes + bytes.len() - 12 > MAX_MESSAGE_BYTES
370            || self.bytes + bytes.len() - 12 > MAX_PENDING_BYTES
371        {
372            self.drop_pending(key);
373            self.remember(key, now);
374            bail!("chunk reassembly byte limit");
375        }
376        let pending = self.pending.get_mut(&key).unwrap();
377        pending.parts[seq] = Some(bytes[12..].to_vec());
378        pending.bytes += bytes.len() - 12;
379        self.bytes += bytes.len() - 12;
380        if pending.parts.iter().any(Option::is_none) {
381            return Ok(None);
382        }
383        let pending = self.pending.remove(&key).unwrap();
384        self.bytes -= pending.bytes;
385        self.remember(key, now);
386        let all: Vec<u8> = pending.parts.into_iter().flat_map(|p| p.unwrap()).collect();
387        decompress(&all).map(Some)
388    }
389}
390#[derive(Default)]
391pub struct TcpDecoder {
392    bytes: Vec<u8>,
393}
394impl TcpDecoder {
395    pub fn feed(&mut self, bytes: &[u8]) -> Result<()> {
396        ensure!(
397            self.bytes.len() + bytes.len() <= MAX_MESSAGE_BYTES + 8192,
398            "TCP buffered byte limit"
399        );
400        self.bytes.extend_from_slice(bytes);
401        Ok(())
402    }
403    pub fn next_message(&mut self) -> Result<Option<Message>> {
404        if let Some(end) = self.bytes.iter().position(|b| *b == 0) {
405            ensure!(end <= MAX_MESSAGE_BYTES, "TCP message byte limit");
406            let message = parse_json(&self.bytes[..end])?;
407            self.bytes.drain(..=end);
408            return Ok(Some(message));
409        }
410        ensure!(
411            self.bytes.len() <= MAX_MESSAGE_BYTES,
412            "TCP message byte limit"
413        );
414        Ok(None)
415    }
416    pub fn finish(&self) -> Result<()> {
417        ensure!(self.bytes.is_empty(), "EOF within GELF TCP frame");
418        Ok(())
419    }
420}
netget/server/gelf/codec.rs

netget/server/gelf/
codec.rs
