netget/server/zabbix/
wire.rs

1//! The Zabbix trapper wire format: the `ZBXD` header, the `sender data` request, and the
2//! response `zabbix_sender` parses.
3//!
4//! Everything here is a pure function of its arguments, shared by the session loop, the action
5//! executor, the tests and the `zabbix_packet` fuzz target. **The model never writes the
6//! response**: it supplies two counts, and [`render_result`] writes the JSON and the `info`
7//! string `processed: P; failed: F; total: T; seconds spent: S` that `zabbix_sender` scans with
8//! `sscanf` to choose its exit status.
9//!
10//! The header is Zabbix's "Zabbix protocol" framing (Zabbix 4.0+):
11//!
12//! ```text
13//! "ZBXD" | flags (1) | data length | reserved
14//!                      4 bytes LE    4 bytes LE      (flags without 0x04)
15//!                      8 bytes LE    8 bytes LE      (flags with 0x04, "large packet")
16//! ```
17//!
18//! `0x01` must be set; `0x02` means the data is zlib-compressed and `reserved` is its
19//! uncompressed size. NetGet refuses compressed packets: `zabbix_sender` 7.4 does not compress
20//! (measured — it sends flags `0x01`), and refusing removes the one path where the bytes read
21//! and the bytes processed differ.
22
23use serde_json::Value;
24
25pub const MAGIC: &[u8; 4] = b"ZBXD";
26pub const FLAG_PROTOCOL: u8 = 0x01;
27pub const FLAG_COMPRESSED: u8 = 0x02;
28pub const FLAG_LARGE: u8 = 0x04;
29
30/// Header length without and with [`FLAG_LARGE`].
31pub const HEADER_LEN: usize = 13;
32pub const LARGE_HEADER_LEN: usize = 21;
33
34/// The largest request body NetGet reads: 1 MiB.
35///
36/// The protocol permits 1 GiB (Zabbix's `ZBX_MAX_RECV_DATA_SIZE`), which is a limit for a
37/// server that streams values into a database. Every byte here goes into one model prompt as
38/// the event's item list, and `zabbix_sender` splits its input into requests of at most 250
39/// values, so 1 MiB is ~4 KiB per value at the sender's own batch size — far past a metric, a
40/// log line or a short text item — while a gigabyte would be a prompt no backend accepts.
41/// Checked against the **declared** length before anything is allocated for it.
42pub const MAX_DATA_BYTES: usize = 1024 * 1024;
43
44/// The most values one request may carry. Four times `zabbix_sender`'s own batch of 250, for
45/// other senders; past it the request is refused rather than handed to the model.
46pub const MAX_ITEMS: usize = 1000;
47
48/// Why a header was refused.
49#[derive(Debug, Clone, PartialEq, Eq)]
50pub enum HeaderError {
51    /// The first four bytes are not `ZBXD`.
52    BadMagic,
53    /// `0x01` is not set, or a bit other than `0x01`/`0x02`/`0x04` is.
54    BadFlags(u8),
55    /// `0x02`: zlib-compressed data, which NetGet does not accept.
56    Compressed,
57    /// The declared data length is past [`MAX_DATA_BYTES`].
58    TooLarge(u64),
59}
60
61#[derive(Debug, Clone, Copy, PartialEq, Eq)]
62pub struct Header {
63    pub flags: u8,
64    pub data_len: u64,
65    pub reserved: u64,
66}
67
68/// How long the header is, given its flags byte.
69pub fn header_len(flags: u8) -> usize {
70    if flags & FLAG_LARGE != 0 {
71        LARGE_HEADER_LEN
72    } else {
73        HEADER_LEN
74    }
75}
76
77/// Parse a complete header (`bytes.len() >= header_len(bytes[4])`). The length is judged here,
78/// before the caller allocates anything for the body.
79pub fn parse_header(bytes: &[u8]) -> Result<Header, HeaderError> {
80    if bytes.len() < 5 || &bytes[..4] != MAGIC {
81        return Err(HeaderError::BadMagic);
82    }
83    let flags = bytes[4];
84    if flags & FLAG_PROTOCOL == 0 || flags & !(FLAG_PROTOCOL | FLAG_COMPRESSED | FLAG_LARGE) != 0 {
85        return Err(HeaderError::BadFlags(flags));
86    }
87    let need = header_len(flags);
88    if bytes.len() < need {
89        return Err(HeaderError::BadMagic);
90    }
91    let (data_len, reserved) = if flags & FLAG_LARGE != 0 {
92        (
93            u64::from_le_bytes(bytes[5..13].try_into().unwrap()),
94            u64::from_le_bytes(bytes[13..21].try_into().unwrap()),
95        )
96    } else {
97        (
98            u64::from(u32::from_le_bytes(bytes[5..9].try_into().unwrap())),
99            u64::from(u32::from_le_bytes(bytes[9..13].try_into().unwrap())),
100        )
101    };
102    if flags & FLAG_COMPRESSED != 0 {
103        return Err(HeaderError::Compressed);
104    }
105    if data_len > MAX_DATA_BYTES as u64 {
106        return Err(HeaderError::TooLarge(data_len));
107    }
108    Ok(Header {
109        flags,
110        data_len,
111        reserved,
112    })
113}
114
115/// Frame `payload` as Zabbix does for an uncompressed packet that fits the standard header.
116pub fn encode(payload: &[u8]) -> Vec<u8> {
117    let mut out = Vec::with_capacity(HEADER_LEN + payload.len());
118    out.extend_from_slice(MAGIC);
119    out.push(FLAG_PROTOCOL);
120    out.extend_from_slice(&(payload.len() as u32).to_le_bytes());
121    out.extend_from_slice(&0u32.to_le_bytes());
122    out.extend_from_slice(payload);
123    out
124}
125
126/// Frame `payload` with the large (8-byte length) header.
127pub fn encode_large(payload: &[u8]) -> Vec<u8> {
128    let mut out = Vec::with_capacity(LARGE_HEADER_LEN + payload.len());
129    out.extend_from_slice(MAGIC);
130    out.push(FLAG_PROTOCOL | FLAG_LARGE);
131    out.extend_from_slice(&(payload.len() as u64).to_le_bytes());
132    out.extend_from_slice(&0u64.to_le_bytes());
133    out.extend_from_slice(payload);
134    out
135}
136
137/// One value a sender reported.
138#[derive(Debug, Clone, PartialEq, Eq)]
139pub struct SenderItem {
140    pub host: String,
141    pub key: String,
142    pub value: String,
143    pub clock: Option<i64>,
144    pub ns: Option<i64>,
145}
146
147/// A parsed request body.
148#[derive(Debug, Clone, PartialEq, Eq)]
149pub enum Request {
150    SenderData {
151        items: Vec<SenderItem>,
152        clock: Option<i64>,
153    },
154    /// A well-formed request of a kind this server does not answer (`active checks`,
155    /// `agent data`, `zabbix.stats`, …), named.
156    Other(String),
157}
158
159/// Why a request body was refused. The texts are fixed and go to the peer in `info`.
160#[derive(Debug, Clone, Copy, PartialEq, Eq)]
161pub enum RequestError {
162    NotJson,
163    NoRequest,
164    BadData,
165    TooManyItems,
166}
167
168impl RequestError {
169    pub fn info(self) -> &'static str {
170        match self {
171            RequestError::NotJson => "cannot parse request as a JSON object",
172            RequestError::NoRequest => "cannot find the \"request\" tag",
173            RequestError::BadData => "cannot parse the \"data\" array",
174            RequestError::TooManyItems => "too many values in one request",
175        }
176    }
177}
178
179fn text(v: Option<&Value>) -> Option<String> {
180    match v? {
181        Value::String(s) => Some(s.clone()),
182        Value::Number(n) => Some(n.to_string()),
183        Value::Bool(b) => Some(b.to_string()),
184        _ => None,
185    }
186}
187
188fn int(v: Option<&Value>) -> Option<i64> {
189    match v? {
190        Value::Number(n) => n.as_i64(),
191        Value::String(s) => s.trim().parse().ok(),
192        _ => None,
193    }
194}
195
196/// Parse a request body. serde_json bounds nesting (128 levels) on its own, so a depth bomb is
197/// a parse error rather than a stack overflow.
198pub fn parse_request(data: &[u8]) -> Result<Request, RequestError> {
199    let json: Value = serde_json::from_slice(data).map_err(|_| RequestError::NotJson)?;
200    let obj = json.as_object().ok_or(RequestError::NotJson)?;
201    let request = obj
202        .get("request")
203        .and_then(Value::as_str)
204        .ok_or(RequestError::NoRequest)?;
205    if request != "sender data" {
206        return Ok(Request::Other(request.to_string()));
207    }
208    let data = obj
209        .get("data")
210        .and_then(Value::as_array)
211        .ok_or(RequestError::BadData)?;
212    if data.len() > MAX_ITEMS {
213        return Err(RequestError::TooManyItems);
214    }
215    let mut items = Vec::with_capacity(data.len());
216    for entry in data {
217        let entry = entry.as_object().ok_or(RequestError::BadData)?;
218        items.push(SenderItem {
219            host: text(entry.get("host")).ok_or(RequestError::BadData)?,
220            key: text(entry.get("key")).ok_or(RequestError::BadData)?,
221            value: text(entry.get("value")).unwrap_or_default(),
222            clock: int(entry.get("clock")),
223            ns: int(entry.get("ns")),
224        });
225    }
226    Ok(Request::SenderData {
227        items,
228        clock: int(obj.get("clock")),
229    })
230}
231
232/// `processed: P; failed: F; total: T; seconds spent: S` — the string zabbix_sender scans.
233pub fn info_string(processed: u64, failed: u64, total: u64, seconds: f64) -> String {
234    format!("processed: {processed}; failed: {failed}; total: {total}; seconds spent: {seconds:.6}")
235}
236
237/// A complete `success` response packet.
238pub fn render_result(processed: u64, failed: u64, total: u64, seconds: f64) -> Vec<u8> {
239    let body = serde_json::json!({
240        "response": "success",
241        "info": info_string(processed, failed, total, seconds),
242    });
243    encode(body.to_string().as_bytes())
244}
245
246/// A complete `failed` response packet. `info` is always one of this module's fixed texts.
247pub fn render_failed(info: &'static str) -> Vec<u8> {
248    let body = serde_json::json!({"response": "failed", "info": info});
249    encode(body.to_string().as_bytes())
250}
251
252/// Read back what [`render_result`] wrote: `(processed, failed, total)`, or `None` for anything
253/// else. The session loop uses this to check a reply against the request it answers.
254pub fn read_result(packet: &[u8]) -> Option<(u64, u64, u64)> {
255    let header = parse_header(packet).ok()?;
256    let body = packet.get(header_len(header.flags)..)?;
257    let json: Value = serde_json::from_slice(body).ok()?;
258    if json.get("response")?.as_str()? != "success" {
259        return None;
260    }
261    let info = json.get("info")?.as_str()?;
262    let mut fields = info.split("; ");
263    let mut next = |name: &str| -> Option<u64> {
264        fields
265            .next()?
266            .strip_prefix(name)?
267            .strip_prefix(": ")?
268            .parse()
269            .ok()
270    };
271    Some((next("processed")?, next("failed")?, next("total")?))
272}
netget/server/zabbix/wire.rs

netget/server/zabbix/
wire.rs
