netget/server/amqp1/
types.rs

1//! The AMQP 1.0 type system (OASIS AMQP 1.0, part 1): primitive encodings, lists, maps, arrays
2//! and described types, decoded with depth and size bounds.
3use anyhow::{bail, ensure, Context, Result};
4
5const MAX_DEPTH: usize = 32;
6const MAX_ITEMS: usize = 100_000;
7
8#[derive(Debug, Clone, PartialEq)]
9pub enum Value {
10    Null,
11    Bool(bool),
12    Ubyte(u8),
13    Ushort(u16),
14    Uint(u32),
15    Ulong(u64),
16    Byte(i8),
17    Short(i16),
18    Int(i32),
19    Long(i64),
20    Float(f32),
21    Double(f64),
22    Char(char),
23    Timestamp(i64),
24    Uuid([u8; 16]),
25    Binary(Vec<u8>),
26    String(String),
27    Symbol(String),
28    Decimal(Vec<u8>),
29    List(Vec<Value>),
30    Map(Vec<(Value, Value)>),
31    Array(Vec<Value>),
32    Described(Box<Value>, Box<Value>),
33}
34
35impl Value {
36    pub fn as_u64(&self) -> Option<u64> {
37        match self {
38            Value::Ubyte(n) => Some(*n as u64),
39            Value::Ushort(n) => Some(*n as u64),
40            Value::Uint(n) => Some(*n as u64),
41            Value::Ulong(n) => Some(*n),
42            Value::Byte(n) if *n >= 0 => Some(*n as u64),
43            Value::Short(n) if *n >= 0 => Some(*n as u64),
44            Value::Int(n) if *n >= 0 => Some(*n as u64),
45            Value::Long(n) if *n >= 0 => Some(*n as u64),
46            _ => None,
47        }
48    }
49    pub fn as_str(&self) -> Option<&str> {
50        match self {
51            Value::String(s) | Value::Symbol(s) => Some(s),
52            _ => None,
53        }
54    }
55    pub fn as_bool(&self) -> Option<bool> {
56        match self {
57            Value::Bool(b) => Some(*b),
58            _ => None,
59        }
60    }
61    pub fn is_null(&self) -> bool {
62        matches!(self, Value::Null)
63    }
64    /// A described value's descriptor as a numeric code (symbolic descriptors mapped).
65    pub fn descriptor(&self) -> Option<u64> {
66        match self {
67            Value::Described(d, _) => d.as_u64().or_else(|| d.as_str().and_then(symbolic_code)),
68            _ => None,
69        }
70    }
71    /// The fields of a described list (missing trailing fields are absent).
72    pub fn fields(&self) -> &[Value] {
73        match self {
74            Value::Described(_, inner) => match inner.as_ref() {
75                Value::List(f) => f,
76                _ => &[],
77            },
78            _ => &[],
79        }
80    }
81    pub fn described(code: u64, fields: Vec<Value>) -> Value {
82        Value::Described(Box::new(Value::Ulong(code)), Box::new(Value::List(fields)))
83    }
84    pub fn sym(s: &str) -> Value {
85        Value::Symbol(s.to_owned())
86    }
87    pub fn str(s: &str) -> Value {
88        Value::String(s.to_owned())
89    }
90}
91
92/// Field `i` of a described list, or Null.
93pub fn field(v: &Value, i: usize) -> &Value {
94    v.fields().get(i).unwrap_or(&Value::Null)
95}
96
97fn symbolic_code(s: &str) -> Option<u64> {
98    Some(match s {
99        "amqp:open:list" => 0x10,
100        "amqp:begin:list" => 0x11,
101        "amqp:attach:list" => 0x12,
102        "amqp:flow:list" => 0x13,
103        "amqp:transfer:list" => 0x14,
104        "amqp:disposition:list" => 0x15,
105        "amqp:detach:list" => 0x16,
106        "amqp:end:list" => 0x17,
107        "amqp:close:list" => 0x18,
108        "amqp:error:list" => 0x1d,
109        "amqp:received:list" => 0x23,
110        "amqp:accepted:list" => 0x24,
111        "amqp:rejected:list" => 0x25,
112        "amqp:released:list" => 0x26,
113        "amqp:modified:list" => 0x27,
114        "amqp:source:list" => 0x28,
115        "amqp:target:list" => 0x29,
116        "amqp:sasl-mechanisms:list" => 0x40,
117        "amqp:sasl-init:list" => 0x41,
118        "amqp:sasl-outcome:list" => 0x44,
119        "amqp:header:list" => 0x70,
120        "amqp:delivery-annotations:map" => 0x71,
121        "amqp:message-annotations:map" => 0x72,
122        "amqp:properties:list" => 0x73,
123        "amqp:application-properties:map" => 0x74,
124        "amqp:data:binary" => 0x75,
125        "amqp:amqp-sequence:list" => 0x76,
126        "amqp:amqp-value:*" => 0x77,
127        "amqp:footer:map" => 0x78,
128        _ => return None,
129    })
130}
131
132struct Reader<'a> {
133    b: &'a [u8],
134    pos: usize,
135    items: usize,
136}
137
138impl Reader<'_> {
139    fn take(&mut self, n: usize) -> Result<&[u8]> {
140        let v = self
141            .b
142            .get(self.pos..self.pos.checked_add(n).context("length overflow")?)
143            .context("truncated AMQP value")?;
144        self.pos += n;
145        Ok(v)
146    }
147    fn u8(&mut self) -> Result<u8> {
148        Ok(self.take(1)?[0])
149    }
150    fn u32(&mut self) -> Result<u32> {
151        Ok(u32::from_be_bytes(self.take(4)?.try_into()?))
152    }
153    fn arr<const N: usize>(&mut self) -> Result<[u8; N]> {
154        Ok(self.take(N)?.try_into()?)
155    }
156
157    fn value(&mut self, depth: usize) -> Result<Value> {
158        let code = self.u8()?;
159        self.with_code(code, depth)
160    }
161
162    fn with_code(&mut self, code: u8, depth: usize) -> Result<Value> {
163        ensure!(depth <= MAX_DEPTH, "AMQP value nests too deeply");
164        self.items += 1;
165        ensure!(self.items <= MAX_ITEMS, "too many AMQP values");
166        Ok(match code {
167            0x00 => {
168                let d = self.value(depth + 1)?;
169                let v = self.value(depth + 1)?;
170                Value::Described(Box::new(d), Box::new(v))
171            }
172            0x40 => Value::Null,
173            0x41 => Value::Bool(true),
174            0x42 => Value::Bool(false),
175            0x56 => Value::Bool(self.u8()? != 0),
176            0x50 => Value::Ubyte(self.u8()?),
177            0x60 => Value::Ushort(u16::from_be_bytes(self.arr()?)),
178            0x70 => Value::Uint(self.u32()?),
179            0x52 => Value::Uint(self.u8()? as u32),
180            0x43 => Value::Uint(0),
181            0x80 => Value::Ulong(u64::from_be_bytes(self.arr()?)),
182            0x53 => Value::Ulong(self.u8()? as u64),
183            0x44 => Value::Ulong(0),
184            0x51 => Value::Byte(self.u8()? as i8),
185            0x61 => Value::Short(i16::from_be_bytes(self.arr()?)),
186            0x71 => Value::Int(i32::from_be_bytes(self.arr()?)),
187            0x54 => Value::Int(self.u8()? as i8 as i32),
188            0x81 => Value::Long(i64::from_be_bytes(self.arr()?)),
189            0x55 => Value::Long(self.u8()? as i8 as i64),
190            0x72 => Value::Float(f32::from_be_bytes(self.arr()?)),
191            0x82 => Value::Double(f64::from_be_bytes(self.arr()?)),
192            0x74 => Value::Decimal(self.take(4)?.to_vec()),
193            0x84 => Value::Decimal(self.take(8)?.to_vec()),
194            0x94 => Value::Decimal(self.take(16)?.to_vec()),
195            0x73 => Value::Char(char::from_u32(self.u32()?).context("invalid char")?),
196            0x83 => Value::Timestamp(i64::from_be_bytes(self.arr()?)),
197            0x98 => Value::Uuid(self.arr()?),
198            0xa0 | 0xb0 => {
199                let n = if code == 0xa0 {
200                    self.u8()? as usize
201                } else {
202                    self.u32()? as usize
203                };
204                Value::Binary(self.take(n)?.to_vec())
205            }
206            0xa1 | 0xb1 | 0xa3 | 0xb3 => {
207                let n = if code & 0xf0 == 0xa0 {
208                    self.u8()? as usize
209                } else {
210                    self.u32()? as usize
211                };
212                let s = std::str::from_utf8(self.take(n)?)
213                    .context("string is not UTF-8")?
214                    .to_owned();
215                if code & 0x0f == 1 {
216                    Value::String(s)
217                } else {
218                    Value::Symbol(s)
219                }
220            }
221            0x45 => Value::List(vec![]),
222            0xc0 | 0xd0 | 0xc1 | 0xd1 => {
223                let wide = code & 0xf0 == 0xd0;
224                let size = if wide {
225                    self.u32()? as usize
226                } else {
227                    self.u8()? as usize
228                };
229                let end = self
230                    .pos
231                    .checked_add(size)
232                    .filter(|e| *e <= self.b.len())
233                    .context("compound longer than its buffer")?;
234                let count = if wide {
235                    self.u32()? as usize
236                } else {
237                    self.u8()? as usize
238                };
239                ensure!(count <= MAX_ITEMS, "compound count too large");
240                let mut items = Vec::with_capacity(count.min(1024));
241                for _ in 0..count {
242                    items.push(self.value(depth + 1)?);
243                }
244                ensure!(self.pos == end, "compound size does not match its contents");
245                if code & 0x0f == 0 {
246                    Value::List(items)
247                } else {
248                    ensure!(count % 2 == 0, "map with an odd count");
249                    let mut m = Vec::with_capacity(items.len() / 2);
250                    let mut it = items.into_iter();
251                    while let (Some(k), Some(v)) = (it.next(), it.next()) {
252                        m.push((k, v));
253                    }
254                    Value::Map(m)
255                }
256            }
257            0xe0 | 0xf0 => {
258                let wide = code == 0xf0;
259                let size = if wide {
260                    self.u32()? as usize
261                } else {
262                    self.u8()? as usize
263                };
264                let end = self
265                    .pos
266                    .checked_add(size)
267                    .filter(|e| *e <= self.b.len())
268                    .context("array longer than its buffer")?;
269                let count = if wide {
270                    self.u32()? as usize
271                } else {
272                    self.u8()? as usize
273                };
274                ensure!(count <= MAX_ITEMS, "array count too large");
275                let mut ctor = self.u8()?;
276                let mut descriptor = None;
277                if ctor == 0x00 {
278                    descriptor = Some(self.value(depth + 1)?);
279                    ctor = self.u8()?;
280                }
281                let mut items = Vec::with_capacity(count.min(1024));
282                for _ in 0..count {
283                    let v = self.with_code(ctor, depth + 1)?;
284                    items.push(match &descriptor {
285                        Some(d) => Value::Described(Box::new(d.clone()), Box::new(v)),
286                        None => v,
287                    });
288                }
289                ensure!(self.pos == end, "array size does not match its contents");
290                Value::Array(items)
291            }
292            other => bail!("unknown AMQP type code 0x{other:02x}"),
293        })
294    }
295}
296
297/// Decode one value from the front of `b`; the value and the bytes it used.
298pub fn decode(b: &[u8]) -> Result<(Value, usize)> {
299    let mut r = Reader {
300        b,
301        pos: 0,
302        items: 0,
303    };
304    let v = r.value(0)?;
305    Ok((v, r.pos))
306}
307
308/// Decode every value in `b` (message sections follow one another).
309pub fn decode_all(b: &[u8]) -> Result<Vec<Value>> {
310    let mut r = Reader {
311        b,
312        pos: 0,
313        items: 0,
314    };
315    let mut out = Vec::new();
316    while r.pos < b.len() {
317        out.push(r.value(0)?);
318    }
319    Ok(out)
320}
321
322fn sized(out: &mut Vec<u8>, short: u8, long: u8, body: &[u8], count: usize) {
323    if body.len() < 255 && count < 256 {
324        out.push(short);
325        out.push((body.len() + 1) as u8);
326        out.push(count as u8);
327    } else {
328        out.push(long);
329        out.extend(((body.len() + 4) as u32).to_be_bytes());
330        out.extend((count as u32).to_be_bytes());
331    }
332    out.extend(body);
333}
334
335/// The constructor code and encoded body of a value, for array elements (one constructor each).
336fn element(v: &Value) -> (u8, Vec<u8>) {
337    let mut full = Vec::new();
338    encode_into(v, &mut full);
339    match v {
340        // Arrays need a fixed constructor: use the wide forms for variable sizes.
341        Value::String(s) => (
342            0xb1,
343            [&(s.len() as u32).to_be_bytes()[..], s.as_bytes()].concat(),
344        ),
345        Value::Symbol(s) => (
346            0xb3,
347            [&(s.len() as u32).to_be_bytes()[..], s.as_bytes()].concat(),
348        ),
349        Value::Binary(b) => (0xb0, [&(b.len() as u32).to_be_bytes()[..], b].concat()),
350        Value::Uint(n) => (0x70, n.to_be_bytes().to_vec()),
351        Value::Ulong(n) => (0x80, n.to_be_bytes().to_vec()),
352        Value::Int(n) => (0x71, n.to_be_bytes().to_vec()),
353        Value::Long(n) => (0x81, n.to_be_bytes().to_vec()),
354        Value::Bool(b) => (0x56, vec![*b as u8]),
355        _ => (full[0], full[1..].to_vec()),
356    }
357}
358
359pub fn encode_into(v: &Value, out: &mut Vec<u8>) {
360    match v {
361        Value::Null => out.push(0x40),
362        Value::Bool(true) => out.push(0x41),
363        Value::Bool(false) => out.push(0x42),
364        Value::Ubyte(n) => out.extend([0x50, *n]),
365        Value::Ushort(n) => {
366            out.push(0x60);
367            out.extend(n.to_be_bytes());
368        }
369        Value::Uint(0) => out.push(0x43),
370        Value::Uint(n) if *n < 256 => out.extend([0x52, *n as u8]),
371        Value::Uint(n) => {
372            out.push(0x70);
373            out.extend(n.to_be_bytes());
374        }
375        Value::Ulong(0) => out.push(0x44),
376        Value::Ulong(n) if *n < 256 => out.extend([0x53, *n as u8]),
377        Value::Ulong(n) => {
378            out.push(0x80);
379            out.extend(n.to_be_bytes());
380        }
381        Value::Byte(n) => out.extend([0x51, *n as u8]),
382        Value::Short(n) => {
383            out.push(0x61);
384            out.extend(n.to_be_bytes());
385        }
386        Value::Int(n) if (-128..128).contains(n) => out.extend([0x54, *n as i8 as u8]),
387        Value::Int(n) => {
388            out.push(0x71);
389            out.extend(n.to_be_bytes());
390        }
391        Value::Long(n) if (-128..128).contains(n) => out.extend([0x55, *n as i8 as u8]),
392        Value::Long(n) => {
393            out.push(0x81);
394            out.extend(n.to_be_bytes());
395        }
396        Value::Float(f) => {
397            out.push(0x72);
398            out.extend(f.to_be_bytes());
399        }
400        Value::Double(f) => {
401            out.push(0x82);
402            out.extend(f.to_be_bytes());
403        }
404        Value::Decimal(b) => {
405            out.push(match b.len() {
406                4 => 0x74,
407                8 => 0x84,
408                _ => 0x94,
409            });
410            out.extend(b);
411        }
412        Value::Char(c) => {
413            out.push(0x73);
414            out.extend((*c as u32).to_be_bytes());
415        }
416        Value::Timestamp(t) => {
417            out.push(0x83);
418            out.extend(t.to_be_bytes());
419        }
420        Value::Uuid(u) => {
421            out.push(0x98);
422            out.extend(u);
423        }
424        Value::Binary(b) => var(out, 0xa0, 0xb0, b),
425        Value::String(s) => var(out, 0xa1, 0xb1, s.as_bytes()),
426        Value::Symbol(s) => var(out, 0xa3, 0xb3, s.as_bytes()),
427        Value::List(items) if items.is_empty() => out.push(0x45),
428        Value::List(items) => {
429            let mut body = Vec::new();
430            for i in items {
431                encode_into(i, &mut body);
432            }
433            sized(out, 0xc0, 0xd0, &body, items.len());
434        }
435        Value::Map(pairs) => {
436            let mut body = Vec::new();
437            for (k, v) in pairs {
438                encode_into(k, &mut body);
439                encode_into(v, &mut body);
440            }
441            sized(out, 0xc1, 0xd1, &body, pairs.len() * 2);
442        }
443        Value::Array(items) => {
444            let (ctor, mut body) = match items.first() {
445                None => (0x40, Vec::new()),
446                Some(first) => (element(first).0, Vec::new()),
447            };
448            for i in items {
449                body.extend(element(i).1);
450            }
451            let mut all = vec![ctor];
452            all.extend(body);
453            // array8/array32: size covers count and constructor and elements.
454            if all.len() < 255 && items.len() < 256 {
455                out.push(0xe0);
456                out.push((all.len() + 1) as u8);
457                out.push(items.len() as u8);
458            } else {
459                out.push(0xf0);
460                out.extend(((all.len() + 4) as u32).to_be_bytes());
461                out.extend((items.len() as u32).to_be_bytes());
462            }
463            out.extend(all);
464        }
465        Value::Described(d, v) => {
466            out.push(0x00);
467            encode_into(d, out);
468            encode_into(v, out);
469        }
470    }
471}
472
473fn var(out: &mut Vec<u8>, short: u8, long: u8, b: &[u8]) {
474    if b.len() < 256 {
475        out.extend([short, b.len() as u8]);
476    } else {
477        out.push(long);
478        out.extend((b.len() as u32).to_be_bytes());
479    }
480    out.extend(b);
481}
482
483pub fn encode(v: &Value) -> Vec<u8> {
484    let mut out = Vec::new();
485    encode_into(v, &mut out);
486    out
487}
netget/server/amqp1/types.rs

netget/server/amqp1/
types.rs
