netget/server/bmp/
codec.rs

1//! BMP version 3 (RFC 7854, with the RFC 8671 Adj-RIB-Out flag and RFC 9069 Loc-RIB peers):
2//! message framing, the per-peer header, TLVs, and the JSON the handler sees. Embedded BGP PDUs
3//! are framed by their own length and decoded by `crate::server::bgp::wire` (netgauze).
4use crate::server::bgp::wire;
5use anyhow::{bail, ensure, Context, Result};
6use netgauze_bgp_pkt::BgpMessage;
7use serde_json::{json, Value as Json};
8use std::net::{IpAddr, Ipv4Addr, Ipv6Addr};
9
10pub const VERSION: u8 = 3;
11pub const HEADER_LEN: usize = 6;
12pub const PEER_HEADER_LEN: usize = 42;
13/// A BMP message carries at most two BGP PDUs (Peer Up) plus TLVs; BGP PDUs are 4096 bytes
14/// unless extended messages were negotiated (65535).
15pub const MAX_MESSAGE: usize = 256 * 1024;
16/// Statistics entries and TLVs read from one message.
17pub const MAX_ITEMS: usize = 1024;
18
19pub const ROUTE_MONITORING: u8 = 0;
20pub const STATISTICS: u8 = 1;
21pub const PEER_DOWN: u8 = 2;
22pub const PEER_UP: u8 = 3;
23pub const INITIATION: u8 = 4;
24pub const TERMINATION: u8 = 5;
25pub const ROUTE_MIRRORING: u8 = 6;
26
27pub const FLAG_V6: u8 = 0x80;
28pub const FLAG_POST_POLICY: u8 = 0x40;
29pub const FLAG_LEGACY_AS: u8 = 0x20;
30pub const FLAG_ADJ_RIB_OUT: u8 = 0x10;
31
32/// The length of the message at the front of `buf`, once its common header is there.
33pub fn message_len(buf: &[u8]) -> Result<Option<usize>> {
34    if buf.len() < HEADER_LEN {
35        return Ok(None);
36    }
37    ensure!(
38        buf[0] == VERSION,
39        "BMP version {} is not supported (only 3)",
40        buf[0]
41    );
42    let len = u32::from_be_bytes(buf[1..5].try_into()?) as usize;
43    ensure!(
44        len >= HEADER_LEN,
45        "BMP message length {len} is below the header"
46    );
47    ensure!(
48        len <= MAX_MESSAGE,
49        "BMP message of {len} bytes exceeds the {MAX_MESSAGE}-byte bound"
50    );
51    Ok(Some(len))
52}
53
54/// Frame a message body with the common header.
55pub fn frame(kind: u8, body: &[u8]) -> Vec<u8> {
56    let mut out = Vec::with_capacity(HEADER_LEN + body.len());
57    out.push(VERSION);
58    out.extend(((HEADER_LEN + body.len()) as u32).to_be_bytes());
59    out.push(kind);
60    out.extend(body);
61    out
62}
63
64#[derive(Clone, Debug, PartialEq)]
65pub struct PeerHeader {
66    pub peer_type: u8,
67    pub flags: u8,
68    pub distinguisher: [u8; 8],
69    pub address: IpAddr,
70    pub asn: u32,
71    pub bgp_id: Ipv4Addr,
72    pub seconds: u32,
73    pub micros: u32,
74}
75
76fn address(b: &[u8], v6: bool) -> IpAddr {
77    let raw: [u8; 16] = b[..16].try_into().unwrap_or([0; 16]);
78    if v6 {
79        IpAddr::V6(Ipv6Addr::from(raw))
80    } else {
81        IpAddr::V4(Ipv4Addr::new(raw[12], raw[13], raw[14], raw[15]))
82    }
83}
84
85fn put_address(out: &mut Vec<u8>, a: IpAddr) {
86    match a {
87        IpAddr::V4(v4) => {
88            out.extend([0u8; 12]);
89            out.extend(v4.octets());
90        }
91        IpAddr::V6(v6) => out.extend(v6.octets()),
92    }
93}
94
95impl PeerHeader {
96    pub fn read(b: &[u8]) -> Result<Self> {
97        ensure!(
98            b.len() >= PEER_HEADER_LEN,
99            "the per-peer header is truncated"
100        );
101        let flags = b[1];
102        Ok(Self {
103            peer_type: b[0],
104            flags,
105            distinguisher: b[2..10].try_into()?,
106            address: address(&b[10..26], flags & FLAG_V6 != 0),
107            asn: u32::from_be_bytes(b[26..30].try_into()?),
108            bgp_id: Ipv4Addr::from(<[u8; 4]>::try_from(&b[30..34])?),
109            seconds: u32::from_be_bytes(b[34..38].try_into()?),
110            micros: u32::from_be_bytes(b[38..42].try_into()?),
111        })
112    }
113
114    pub fn write(&self, out: &mut Vec<u8>) {
115        out.push(self.peer_type);
116        let v6 = if self.address.is_ipv6() { FLAG_V6 } else { 0 };
117        out.push((self.flags & !FLAG_V6) | v6);
118        out.extend(self.distinguisher);
119        put_address(out, self.address);
120        out.extend(self.asn.to_be_bytes());
121        out.extend(self.bgp_id.octets());
122        out.extend(self.seconds.to_be_bytes());
123        out.extend(self.micros.to_be_bytes());
124    }
125
126    pub fn four_octet_as(&self) -> bool {
127        self.flags & FLAG_LEGACY_AS == 0
128    }
129
130    pub fn to_json(&self) -> Json {
131        let timestamp = (self.seconds != 0)
132            .then(|| {
133                chrono::DateTime::from_timestamp(
134                    self.seconds as i64,
135                    self.micros.min(999_999) * 1000,
136                )
137                .map(|t| t.to_rfc3339_opts(chrono::SecondsFormat::Micros, true))
138            })
139            .flatten();
140        json!({
141            "type": peer_type_name(self.peer_type),
142            "address": self.address.to_string(),
143            "asn": self.asn,
144            "bgp_id": self.bgp_id.to_string(),
145            "distinguisher": distinguisher(&self.distinguisher),
146            "post_policy": self.flags & FLAG_POST_POLICY != 0,
147            "adj_rib_out": self.flags & FLAG_ADJ_RIB_OUT != 0,
148            "four_octet_as": self.four_octet_as(),
149            "timestamp": timestamp,
150        })
151    }
152}
153
154pub fn peer_type_name(t: u8) -> String {
155    match t {
156        0 => "global".into(),
157        1 => "rd_instance".into(),
158        2 => "local_instance".into(),
159        3 => "loc_rib".into(),
160        n => format!("type_{n}"),
161    }
162}
163
164pub fn peer_type_code(name: &str) -> Option<u8> {
165    Some(match name {
166        "global" => 0,
167        "rd_instance" => 1,
168        "local_instance" => 2,
169        "loc_rib" => 3,
170        _ => return None,
171    })
172}
173
174/// RFC 4364 route distinguisher text: type 0 `asn:n`, type 1 `ip:n`, type 2 `asn4:n`.
175pub fn distinguisher(d: &[u8; 8]) -> String {
176    let t = u16::from_be_bytes([d[0], d[1]]);
177    match t {
178        0 => format!(
179            "{}:{}",
180            u16::from_be_bytes([d[2], d[3]]),
181            u32::from_be_bytes([d[4], d[5], d[6], d[7]])
182        ),
183        1 => format!(
184            "{}:{}",
185            Ipv4Addr::new(d[2], d[3], d[4], d[5]),
186            u16::from_be_bytes([d[6], d[7]])
187        ),
188        2 => format!(
189            "{}:{}",
190            u32::from_be_bytes([d[2], d[3], d[4], d[5]]),
191            u16::from_be_bytes([d[6], d[7]])
192        ),
193        _ => format!("type{t}:{}", hex::encode(&d[2..])),
194    }
195}
196
197pub fn parse_distinguisher(s: &str) -> Result<[u8; 8]> {
198    let (a, n) = s
199        .rsplit_once(':')
200        .context("a route distinguisher is admin:assigned, e.g. 65000:1 or 192.0.2.1:7")?;
201    let mut d = [0u8; 8];
202    if let Ok(ip) = a.parse::<Ipv4Addr>() {
203        d[1] = 1;
204        d[2..6].copy_from_slice(&ip.octets());
205        d[6..8].copy_from_slice(&n.parse::<u16>()?.to_be_bytes());
206    } else {
207        let asn: u32 = a.parse().context("the distinguisher's admin part")?;
208        match u16::try_from(asn) {
209            Ok(a2) => {
210                d[2..4].copy_from_slice(&a2.to_be_bytes());
211                d[4..8].copy_from_slice(&n.parse::<u32>()?.to_be_bytes());
212            }
213            Err(_) => {
214                d[1] = 2;
215                d[2..6].copy_from_slice(&asn.to_be_bytes());
216                d[6..8].copy_from_slice(&n.parse::<u16>()?.to_be_bytes());
217            }
218        }
219    }
220    Ok(d)
221}
222
223/// Type-length-value items with 2-byte type and length.
224pub fn tlvs(mut b: &[u8]) -> Result<Vec<(u16, &[u8])>> {
225    let mut out = Vec::new();
226    while !b.is_empty() {
227        ensure!(b.len() >= 4, "a TLV header is truncated");
228        ensure!(out.len() < MAX_ITEMS, "more than {MAX_ITEMS} TLVs");
229        let t = u16::from_be_bytes([b[0], b[1]]);
230        let l = u16::from_be_bytes([b[2], b[3]]) as usize;
231        ensure!(b.len() >= 4 + l, "a TLV value is truncated");
232        out.push((t, &b[4..4 + l]));
233        b = &b[4 + l..];
234    }
235    Ok(out)
236}
237
238pub fn put_tlv(out: &mut Vec<u8>, t: u16, v: &[u8]) -> Result<()> {
239    let l = u16::try_from(v.len()).context("a TLV value is over 65535 bytes")?;
240    out.extend(t.to_be_bytes());
241    out.extend(l.to_be_bytes());
242    out.extend(v);
243    Ok(())
244}
245
246fn text(v: &[u8]) -> String {
247    String::from_utf8_lossy(v).into_owned()
248}
249
250/// The BGP PDU at the front of `b` and what follows it.
251pub fn bgp_pdu(b: &[u8]) -> Result<(&[u8], &[u8])> {
252    ensure!(b.len() >= wire::BGP_HEADER_LEN, "the BGP PDU is truncated");
253    ensure!(
254        b[..16] == wire::BGP_MARKER,
255        "the BGP PDU's marker is not all ones"
256    );
257    let len = u16::from_be_bytes([b[16], b[17]]) as usize;
258    ensure!(
259        len >= wire::BGP_HEADER_LEN && len <= b.len(),
260        "the BGP PDU's length {len} does not fit the BMP message"
261    );
262    Ok((&b[..len], &b[len..]))
263}
264
265fn open_json(pdu: &[u8]) -> Json {
266    match wire::decode(pdu, true) {
267        Ok(BgpMessage::Open(o)) => json!({
268            "asn": o.my_asn4(),
269            "hold_time": o.hold_time(),
270            "bgp_id": o.bgp_id().to_string(),
271            "capabilities": wire::capabilities_to_json(&o),
272        }),
273        Ok(_) => json!({"decode_error": "not an OPEN message"}),
274        Err(e) => json!({"decode_error": e.to_string()}),
275    }
276}
277
278fn notification_json(pdu: &[u8]) -> Json {
279    if pdu.len() >= wire::BGP_HEADER_LEN + 2 && pdu[18] == wire::MSG_NOTIFICATION {
280        let (code, sub) = (pdu[19], pdu[20]);
281        json!({"code": code, "subcode": sub, "name": wire::error_name(code), "subcode_name": wire::error_subcode_name(code, sub)})
282    } else {
283        json!({"decode_error": "not a NOTIFICATION message"})
284    }
285}
286
287pub fn termination_reason(code: u16) -> &'static str {
288    match code {
289        0 => "administratively_closed",
290        1 => "unspecified",
291        2 => "out_of_resources",
292        3 => "redundant_connection",
293        4 => "permanently_administratively_closed",
294        _ => "unknown",
295    }
296}
297
298pub fn termination_code(name: &str) -> Option<u16> {
299    (0..=4).find(|c| termination_reason(*c) == name)
300}
301
302pub fn peer_down_reason(code: u8) -> &'static str {
303    match code {
304        1 => "local_notification",
305        2 => "local_no_notification",
306        3 => "remote_notification",
307        4 => "remote_no_data",
308        5 => "peer_deconfigured",
309        6 => "local_system_closed",
310        _ => "unknown",
311    }
312}
313
314pub fn peer_down_code(name: &str) -> Option<u8> {
315    (1..=6).find(|c| peer_down_reason(*c) == name)
316}
317
318/// IANA "BMP Statistics Types".
319pub fn stat_name(t: u16) -> String {
320    let name = match t {
321        0 => "rejected_prefixes",
322        1 => "duplicate_prefix_advertisements",
323        2 => "duplicate_withdraws",
324        3 => "invalidated_cluster_list_loop",
325        4 => "invalidated_as_path_loop",
326        5 => "invalidated_originator_id",
327        6 => "invalidated_as_confed_loop",
328        7 => "adj_rib_in_routes",
329        8 => "loc_rib_routes",
330        9 => "adj_rib_in_routes_per_afi_safi",
331        10 => "loc_rib_routes_per_afi_safi",
332        11 => "updates_treated_as_withdraw",
333        12 => "prefixes_treated_as_withdraw",
334        13 => "duplicate_update_messages",
335        14 => "adj_rib_out_pre_policy_routes",
336        15 => "adj_rib_out_post_policy_routes",
337        16 => "adj_rib_out_pre_policy_routes_per_afi_safi",
338        17 => "adj_rib_out_post_policy_routes_per_afi_safi",
339        n => return format!("type_{n}"),
340    };
341    name.into()
342}
343
344pub fn stat_code(name: &str) -> Option<u16> {
345    (0..=17).find(|c| stat_name(*c) == name)
346}
347
348/// Whether a statistics type is a 64-bit gauge (otherwise a 32-bit counter); per-AFI/SAFI
349/// gauges carry AFI and SAFI before the value.
350pub fn stat_shape(t: u16) -> (bool, bool) {
351    match t {
352        7 | 8 | 14 | 15 => (true, false),
353        9 | 10 | 16 | 17 => (true, true),
354        _ => (false, false),
355    }
356}
357
358/// The handler's event for one message: (event id, data). Unknown message types are None.
359pub fn describe(kind: u8, body: &[u8]) -> Result<Option<(&'static str, Json)>> {
360    Ok(Some(match kind {
361        INITIATION | TERMINATION => {
362            let mut data = json!({"strings": []});
363            for (t, v) in tlvs(body)? {
364                match (kind, t) {
365                    (_, 0) => data["strings"].as_array_mut().unwrap().push(json!(text(v))),
366                    (INITIATION, 1) => data["sys_descr"] = json!(text(v)),
367                    (INITIATION, 2) => data["sys_name"] = json!(text(v)),
368                    (TERMINATION, 1) => {
369                        ensure!(v.len() == 2, "a termination reason is two bytes");
370                        let code = u16::from_be_bytes([v[0], v[1]]);
371                        data["reason"] = json!(termination_reason(code));
372                        data["reason_code"] = json!(code);
373                    }
374                    _ => {}
375                }
376            }
377            let id = if kind == INITIATION {
378                "bmp_initiation"
379            } else {
380                "bmp_termination"
381            };
382            (id, data)
383        }
384        ROUTE_MONITORING | STATISTICS | PEER_DOWN | PEER_UP | ROUTE_MIRRORING => {
385            let peer = PeerHeader::read(body)?;
386            let rest = &body[PEER_HEADER_LEN..];
387            let mut data = json!({"peer": peer.to_json()});
388            let id = match kind {
389                ROUTE_MONITORING => {
390                    let (pdu, _) = bgp_pdu(rest)?;
391                    match wire::decode(pdu, peer.four_octet_as()) {
392                        Ok(BgpMessage::Update(u)) => data["update"] = wire::update_to_json(&u),
393                        Ok(_) => data["decode_error"] = json!("the BGP PDU is not an UPDATE"),
394                        Err(e) => data["decode_error"] = json!(e.to_string()),
395                    }
396                    "bmp_route_monitoring"
397                }
398                STATISTICS => {
399                    ensure!(rest.len() >= 4, "the statistics count is truncated");
400                    let mut counters = Vec::new();
401                    for (t, v) in tlvs(&rest[4..])? {
402                        let mut c = json!({"type": stat_name(t), "type_code": t});
403                        match v.len() {
404                            4 => c["value"] = json!(u32::from_be_bytes(v.try_into()?)),
405                            8 => c["value"] = json!(u64::from_be_bytes(v.try_into()?)),
406                            11 => {
407                                c["afi"] = json!(u16::from_be_bytes([v[0], v[1]]));
408                                c["safi"] = json!(v[2]);
409                                c["value"] = json!(u64::from_be_bytes(v[3..11].try_into()?));
410                            }
411                            n => c["value_length"] = json!(n),
412                        }
413                        counters.push(c);
414                    }
415                    data["counters"] = json!(counters);
416                    "bmp_statistics"
417                }
418                PEER_DOWN => {
419                    ensure!(!rest.is_empty(), "the peer down reason is missing");
420                    data["reason"] = json!(peer_down_reason(rest[0]));
421                    data["reason_code"] = json!(rest[0]);
422                    let tail = &rest[1..];
423                    match rest[0] {
424                        1 | 3 => data["notification"] = notification_json(bgp_pdu(tail)?.0),
425                        2 => {
426                            ensure!(tail.len() >= 2, "the FSM event code is truncated");
427                            data["fsm_event"] = json!(u16::from_be_bytes([tail[0], tail[1]]));
428                        }
429                        _ => {}
430                    }
431                    "bmp_peer_down"
432                }
433                PEER_UP => {
434                    ensure!(rest.len() >= 20, "the peer up header is truncated");
435                    data["local_address"] =
436                        json!(address(&rest[..16], peer.flags & FLAG_V6 != 0).to_string());
437                    data["local_port"] = json!(u16::from_be_bytes([rest[16], rest[17]]));
438                    data["remote_port"] = json!(u16::from_be_bytes([rest[18], rest[19]]));
439                    let (sent, after) = bgp_pdu(&rest[20..])?;
440                    let (received, info) = bgp_pdu(after)?;
441                    data["sent_open"] = open_json(sent);
442                    data["received_open"] = open_json(received);
443                    data["information"] = json!(tlvs(info)?
444                        .into_iter()
445                        .map(|(t, v)| json!({"type": t, "value": text(v)}))
446                        .collect::<Vec<_>>());
447                    "bmp_peer_up"
448                }
449                _ => {
450                    let mut messages = Vec::new();
451                    for (t, v) in tlvs(rest)? {
452                        match t {
453                            0 => messages.push(
454                                match wire::decode(bgp_pdu(v)?.0, peer.four_octet_as()) {
455                                    Ok(BgpMessage::Update(u)) => {
456                                        json!({"update": wire::update_to_json(&u)})
457                                    }
458                                    Ok(m) => json!({"bgp_message": format!("{:?}", m.get_type())}),
459                                    Err(e) => json!({"decode_error": e.to_string()}),
460                                },
461                            ),
462                            1 if v.len() == 2 => messages.push(
463                                json!({"information": match u16::from_be_bytes([v[0], v[1]]) {
464                                    0 => "errored_pdu".to_string(),
465                                    1 => "messages_lost".to_string(),
466                                    n => format!("code_{n}"),
467                                }}),
468                            ),
469                            _ => {}
470                        }
471                    }
472                    data["messages"] = json!(messages);
473                    "bmp_route_mirroring"
474                }
475            };
476            (id, data)
477        }
478        _ => return Ok(None),
479    }))
480}
481
482/// A message in its frame, from the exporter's side.
483pub fn initiation(sys_name: &str, sys_descr: &str, strings: &[String]) -> Result<Vec<u8>> {
484    let mut b = Vec::new();
485    put_tlv(&mut b, 2, sys_name.as_bytes())?;
486    put_tlv(&mut b, 1, sys_descr.as_bytes())?;
487    for s in strings {
488        put_tlv(&mut b, 0, s.as_bytes())?;
489    }
490    Ok(frame(INITIATION, &b))
491}
492
493pub fn termination(reason: u16, message: Option<&str>) -> Result<Vec<u8>> {
494    let mut b = Vec::new();
495    if let Some(m) = message {
496        put_tlv(&mut b, 0, m.as_bytes())?;
497    }
498    put_tlv(&mut b, 1, &reason.to_be_bytes())?;
499    Ok(frame(TERMINATION, &b))
500}
501
502pub fn with_peer(kind: u8, peer: &PeerHeader, rest: &[u8]) -> Result<Vec<u8>> {
503    let mut b = Vec::with_capacity(PEER_HEADER_LEN + rest.len());
504    peer.write(&mut b);
505    b.extend(rest);
506    if b.len() + HEADER_LEN > MAX_MESSAGE {
507        bail!("the BMP message would exceed {MAX_MESSAGE} bytes");
508    }
509    Ok(frame(kind, &b))
510}
netget/server/bmp/codec.rs

netget/server/bmp/
codec.rs
