netget/server/sflow/
codec.rs

1//! Bounded sFlow(R) v5 XDR framing. Opaque payloads never reach model data.
2use anyhow::{bail, ensure, Context, Result};
3use serde::{Deserialize, Serialize};
4use std::{
5    collections::BTreeMap,
6    net::{IpAddr, Ipv4Addr, Ipv6Addr, SocketAddr},
7    time::Duration,
8};
9use tokio::time::Instant;
10
11pub const MAX_MESSAGE_BYTES: usize = 8192;
12pub const MAX_SAMPLES: usize = 32;
13pub const MAX_RECORDS_PER_SAMPLE: usize = 64;
14pub const MAX_RECORDS: usize = 256;
15pub const MAX_HEADER_BYTES: usize = 256;
16pub const MAX_OPAQUE_BYTES: usize = 4096;
17pub const MAX_SESSIONS: usize = 128;
18pub const MAX_IPV6_EXTENSIONS: usize = 8;
19pub const DEFAULT_SESSION_IDLE: u64 = 1800;
20pub const DEFAULT_LLM_FALLBACK: bool = false;
21
22#[derive(Clone, Debug, Serialize, Deserialize)]
23#[serde(deny_unknown_fields)]
24pub struct Source {
25    pub class: u32,
26    pub index: u32,
27}
28#[derive(Clone, Debug, Serialize, Deserialize)]
29#[serde(deny_unknown_fields)]
30pub struct Interface {
31    pub format: u32,
32    pub value: u32,
33}
34#[derive(Clone, Debug, Serialize, Deserialize)]
35#[serde(deny_unknown_fields)]
36pub struct Packet {
37    pub source_ip: IpAddr,
38    pub destination_ip: IpAddr,
39    pub packet_length: u32,
40    pub protocol: u8,
41    pub source_port: u16,
42    pub destination_port: u16,
43    pub tcp_flags: u32,
44    pub traffic_class: u8,
45}
46impl Packet {
47    fn validate(&self, ipv6: bool) -> Result<()> {
48        ensure!(
49            self.source_ip.is_ipv6() == ipv6 && self.destination_ip.is_ipv6() == ipv6,
50            "packet address families must match record"
51        );
52        ensure!(self.tcp_flags <= 0x1ff, "TCP flags limit9bits");
53        Ok(())
54    }
55}
56macro_rules! counters {
57    ($name:ident {$($field:ident:$ty:ident),* $(,)?}) => {
58        #[derive(Clone,Debug,Serialize,Deserialize)]
59        #[serde(deny_unknown_fields)]
60        pub struct $name { $(pub $field:$ty),* }
61        impl $name {
62            fn encode(&self,out:&mut Vec<u8>) { $(out.extend_from_slice(&self.$field.to_be_bytes());)* }
63            fn decode(r:&mut Reader<'_>)->Result<Self>{Ok(Self{$($field:r.$ty()?),*})}
64        }
65    };
66}
67counters!(InterfaceCounters {
68    if_index: u32,
69    if_type: u32,
70    if_speed: u64,
71    if_direction: u32,
72    if_status: u32,
73    in_octets: u64,
74    in_unicast_packets: u32,
75    in_multicast_packets: u32,
76    in_broadcast_packets: u32,
77    in_discards: u32,
78    in_errors: u32,
79    in_unknown_protocols: u32,
80    out_octets: u64,
81    out_unicast_packets: u32,
82    out_multicast_packets: u32,
83    out_broadcast_packets: u32,
84    out_discards: u32,
85    out_errors: u32,
86    promiscuous_mode: u32
87});
88counters!(EthernetCounters {
89    alignment_errors: u32,
90    fcs_errors: u32,
91    single_collision_frames: u32,
92    multiple_collision_frames: u32,
93    sqe_test_errors: u32,
94    deferred_transmissions: u32,
95    late_collisions: u32,
96    excessive_collisions: u32,
97    internal_mac_transmit_errors: u32,
98    carrier_sense_errors: u32,
99    frame_too_longs: u32,
100    internal_mac_receive_errors: u32,
101    symbol_errors: u32
102});
103counters!(VlanCounters {
104    vlan_id: u32,
105    octets: u64,
106    unicast_packets: u32,
107    multicast_packets: u32,
108    broadcast_packets: u32,
109    discards: u32
110});
111#[derive(Clone, Debug, Serialize, Deserialize)]
112#[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)]
113pub enum CounterRecord {
114    Interface(InterfaceCounters),
115    Ethernet(EthernetCounters),
116    Vlan(VlanCounters),
117}
118#[derive(Clone, Debug, Serialize, Deserialize)]
119#[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)]
120pub enum FlowRecord {
121    SynthesizedHeader {
122        packet: Packet,
123        frame_length: u32,
124        stripped: u32,
125    },
126    SampledIpv4(Packet),
127    SampledIpv6(Packet),
128    Ethernet {
129        frame_length: u32,
130        source_mac: String,
131        destination_mac: String,
132        ether_type: u32,
133    },
134    ExtendedSwitch {
135        source_vlan: u32,
136        source_priority: u32,
137        destination_vlan: u32,
138        destination_priority: u32,
139    },
140}
141#[derive(Clone, Debug, Serialize, Deserialize)]
142#[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)]
143pub enum Sample {
144    Flow {
145        #[serde(default)]
146        expanded: bool,
147        sequence_number: u32,
148        source: Source,
149        sampling_rate: u32,
150        sample_pool: u32,
151        drops: u32,
152        input: Interface,
153        output: Interface,
154        records: Vec<FlowRecord>,
155    },
156    Counters {
157        #[serde(default)]
158        expanded: bool,
159        sequence_number: u32,
160        source: Source,
161        records: Vec<CounterRecord>,
162    },
163}
164#[derive(Clone, Debug, Serialize, Deserialize)]
165#[serde(deny_unknown_fields)]
166pub struct Batch {
167    pub agent_address: IpAddr,
168    pub sub_agent_id: u32,
169    #[serde(default)]
170    pub uptime_ms: Option<u32>,
171    pub samples: Vec<Sample>,
172}
173#[derive(Debug, Serialize)]
174pub struct LinkLayer {
175    pub source_mac: String,
176    pub destination_mac: String,
177    pub ether_type: u32,
178    pub vlans: Vec<u16>,
179}
180#[derive(Debug, Serialize)]
181#[serde(tag = "kind", rename_all = "snake_case")]
182pub enum Record {
183    SampledHeader {
184        header_protocol: u32,
185        frame_length: u32,
186        stripped: u32,
187        header_length: usize,
188        packet: Option<Packet>,
189        link_layer: Option<LinkLayer>,
190        decode_status: &'static str,
191    },
192    SampledIpv4(Packet),
193    SampledIpv6(Packet),
194    Ethernet {
195        frame_length: u32,
196        source_mac: String,
197        destination_mac: String,
198        ether_type: u32,
199    },
200    ExtendedSwitch {
201        source_vlan: u32,
202        source_priority: u32,
203        destination_vlan: u32,
204        destination_priority: u32,
205    },
206    InterfaceCounters(InterfaceCounters),
207    EthernetCounters(EthernetCounters),
208    VlanCounters(VlanCounters),
209    Unknown {
210        enterprise: u32,
211        format: u32,
212        byte_count: usize,
213    },
214}
215#[derive(Debug, Serialize)]
216pub struct FlowMeta {
217    pub sampling_rate: u32,
218    pub sample_pool: u32,
219    pub drops: u32,
220    pub input: Interface,
221    pub output: Interface,
222}
223#[derive(Debug, Serialize)]
224#[serde(tag = "kind", rename_all = "snake_case")]
225pub enum DecodedSample {
226    Flow {
227        expanded: bool,
228        sequence_number: u32,
229        source: Source,
230        flow: FlowMeta,
231        records: Vec<Record>,
232    },
233    Counters {
234        expanded: bool,
235        sequence_number: u32,
236        source: Source,
237        records: Vec<Record>,
238    },
239    Unknown {
240        enterprise: u32,
241        format: u32,
242        byte_count: usize,
243    },
244}
245#[derive(Debug, Serialize)]
246pub struct Message {
247    pub agent_address: IpAddr,
248    pub sub_agent_id: u32,
249    pub sequence_number: u32,
250    pub uptime_ms: u32,
251    pub samples: Vec<DecodedSample>,
252    pub record_count: usize,
253    pub sequence_tracking: SequenceTracking,
254}
255#[derive(Debug, Serialize)]
256pub struct SequenceTracking {
257    pub expected: Option<u32>,
258    pub status: &'static str,
259    pub missing_datagrams: Option<u32>,
260    pub uptime_decreased: bool,
261}
262struct Reader<'a> {
263    bytes: &'a [u8],
264    at: usize,
265}
266impl<'a> Reader<'a> {
267    fn new(bytes: &'a [u8]) -> Self {
268        Self { bytes, at: 0 }
269    }
270    fn left(&self) -> usize {
271        self.bytes.len() - self.at
272    }
273    fn take(&mut self, n: usize) -> Result<&'a [u8]> {
274        ensure!(n <= self.left(), "truncated sFlow XDR field");
275        let s = &self.bytes[self.at..self.at + n];
276        self.at += n;
277        Ok(s)
278    }
279    fn u32(&mut self) -> Result<u32> {
280        Ok(u32::from_be_bytes(self.take(4)?.try_into()?))
281    }
282    fn u64(&mut self) -> Result<u64> {
283        Ok(u64::from_be_bytes(self.take(8)?.try_into()?))
284    }
285    fn opaque(&mut self, limit: usize) -> Result<&'a [u8]> {
286        let n = self.u32()? as usize;
287        ensure!(n <= limit, "sFlow opaque byte bound");
288        let data = self.take(n)?;
289        self.take((4 - n % 4) % 4)?;
290        Ok(data)
291    }
292    fn done(&self) -> Result<()> {
293        ensure!(self.left() == 0, "unexpected trailing sFlow fields");
294        Ok(())
295    }
296}
297fn int(out: &mut Vec<u8>, n: u32) {
298    out.extend_from_slice(&n.to_be_bytes());
299}
300fn opaque(out: &mut Vec<u8>, tag: u32, data: &[u8]) -> Result<()> {
301    ensure!(
302        data.len() <= MAX_MESSAGE_BYTES,
303        "sFlow structure byte limit"
304    );
305    int(out, tag);
306    int(out, data.len() as u32);
307    out.extend_from_slice(data);
308    out.resize(out.len() + (4 - data.len() % 4) % 4, 0);
309    ensure!(
310        out.len() <= MAX_MESSAGE_BYTES,
311        "sFlow message byte limit8192"
312    );
313    Ok(())
314}
315fn source(out: &mut Vec<u8>, s: &Source, expanded: bool) -> Result<()> {
316    ensure!(s.class <= 2, "unsupported source class");
317    if expanded {
318        int(out, s.class);
319        int(out, s.index);
320    } else {
321        ensure!(s.index < 1 << 24, "compact source index limit24bits");
322        int(out, (s.class << 24) | s.index);
323    }
324    Ok(())
325}
326fn interface(out: &mut Vec<u8>, i: &Interface, expanded: bool) -> Result<()> {
327    ensure!(i.format <= 2, "unsupported interface format");
328    if expanded {
329        int(out, i.format);
330        int(out, i.value);
331    } else {
332        ensure!(i.value < 1 << 30, "compact interface value limit30bits");
333        int(out, (i.format << 30) | i.value);
334    }
335    Ok(())
336}
337fn read_source(r: &mut Reader<'_>, expanded: bool) -> Result<Source> {
338    let n = r.u32()?;
339    Ok(if expanded {
340        Source {
341            class: n,
342            index: r.u32()?,
343        }
344    } else {
345        Source {
346            class: n >> 24,
347            index: n & 0xffffff,
348        }
349    })
350}
351fn read_interface(r: &mut Reader<'_>, expanded: bool) -> Result<Interface> {
352    let n = r.u32()?;
353    Ok(if expanded {
354        Interface {
355            format: n,
356            value: r.u32()?,
357        }
358    } else {
359        Interface {
360            format: n >> 30,
361            value: n & 0x3fffffff,
362        }
363    })
364}
365pub fn encode(batch: &Batch, sequence: u32, uptime: u32) -> Result<(Vec<u8>, usize)> {
366    ensure!(
367        (1..=MAX_SAMPLES).contains(&batch.samples.len()),
368        "sample count must be1..32"
369    );
370    let mut out = Vec::new();
371    int(&mut out, 5);
372    match batch.agent_address {
373        IpAddr::V4(ip) => {
374            int(&mut out, 1);
375            out.extend_from_slice(&ip.octets());
376        }
377        IpAddr::V6(ip) => {
378            int(&mut out, 2);
379            out.extend_from_slice(&ip.octets());
380        }
381    }
382    int(&mut out, batch.sub_agent_id);
383    int(&mut out, sequence);
384    int(&mut out, uptime);
385    int(&mut out, batch.samples.len() as u32);
386    let mut count = 0;
387    for sample in &batch.samples {
388        let mut body = Vec::new();
389        let (tag, n) = match sample {
390            Sample::Flow {
391                expanded,
392                sequence_number,
393                source: s,
394                sampling_rate,
395                sample_pool,
396                drops,
397                input,
398                output,
399                records,
400            } => {
401                ensure!(
402                    (1..=MAX_RECORDS_PER_SAMPLE).contains(&records.len()),
403                    "flow record count1..64"
404                );
405                int(&mut body, *sequence_number);
406                source(&mut body, s, *expanded)?;
407                int(&mut body, *sampling_rate);
408                int(&mut body, *sample_pool);
409                int(&mut body, *drops);
410                interface(&mut body, input, *expanded)?;
411                interface(&mut body, output, *expanded)?;
412                int(&mut body, records.len() as u32);
413                for record in records {
414                    let (t, b) = encode_flow(record)?;
415                    opaque(&mut body, t, &b)?;
416                }
417                (if *expanded { 3 } else { 1 }, records.len())
418            }
419            Sample::Counters {
420                expanded,
421                sequence_number,
422                source: s,
423                records,
424            } => {
425                ensure!(
426                    (1..=MAX_RECORDS_PER_SAMPLE).contains(&records.len()),
427                    "counter record count1..64"
428                );
429                int(&mut body, *sequence_number);
430                source(&mut body, s, *expanded)?;
431                int(&mut body, records.len() as u32);
432                for record in records {
433                    let mut b = Vec::new();
434                    let t = match record {
435                        CounterRecord::Interface(c) => {
436                            ensure!(
437                                c.if_direction <= 4 && c.if_status <= 3 && c.promiscuous_mode <= 2,
438                                "interface counter flags"
439                            );
440                            c.encode(&mut b);
441                            1
442                        }
443                        CounterRecord::Ethernet(c) => {
444                            c.encode(&mut b);
445                            2
446                        }
447                        CounterRecord::Vlan(c) => {
448                            ensure!(c.vlan_id <= 4095, "VLAN identifier limit4095");
449                            c.encode(&mut b);
450                            5
451                        }
452                    };
453                    opaque(&mut body, t, &b)?;
454                }
455                (if *expanded { 4 } else { 2 }, records.len())
456            }
457        };
458        count += n;
459        ensure!(count <= MAX_RECORDS, "total record count limit256");
460        opaque(&mut out, tag, &body)?;
461    }
462    Ok((out, count))
463}
464fn mac(text: &str) -> Result<[u8; 6]> {
465    ensure!(text.len() == 17, "MAC address must contain six octets");
466    let mut out = [0; 6];
467    let parts = text.split(':').collect::<Vec<_>>();
468    ensure!(parts.len() == 6, "MAC address delimiter");
469    for (i, p) in parts.iter().enumerate() {
470        ensure!(p.len() == 2, "MAC octet width");
471        out[i] = u8::from_str_radix(p, 16)?;
472    }
473    Ok(out)
474}
475fn mac_text(b: &[u8]) -> String {
476    b.iter()
477        .map(|n| format!("{n:02x}"))
478        .collect::<Vec<_>>()
479        .join(":")
480}
481fn encode_flow(record: &FlowRecord) -> Result<(u32, Vec<u8>)> {
482    let mut out = Vec::new();
483    let tag = match record {
484        FlowRecord::SynthesizedHeader {
485            packet,
486            frame_length,
487            stripped,
488        } => {
489            let ipv6 = packet.source_ip.is_ipv6();
490            packet.validate(ipv6)?;
491            ensure!(
492                *frame_length >= *stripped,
493                "stripped exceeds original frame"
494            );
495            let b = synthesized_header(packet)?;
496            ensure!(
497                usize::try_from(frame_length - stripped)? >= b.len(),
498                "captured header exceeds remaining frame"
499            );
500            int(&mut out, if ipv6 { 12 } else { 11 });
501            int(&mut out, *frame_length);
502            int(&mut out, *stripped);
503            int(&mut out, b.len() as u32);
504            out.extend_from_slice(&b);
505            out.resize(out.len() + (4 - b.len() % 4) % 4, 0);
506            1
507        }
508        FlowRecord::SampledIpv4(p) | FlowRecord::SampledIpv6(p) => {
509            let ipv6 = matches!(record, FlowRecord::SampledIpv6(_));
510            p.validate(ipv6)?;
511            int(&mut out, p.packet_length);
512            int(&mut out, u32::from(p.protocol));
513            match (p.source_ip, p.destination_ip) {
514                (IpAddr::V4(a), IpAddr::V4(b)) => {
515                    out.extend_from_slice(&a.octets());
516                    out.extend_from_slice(&b.octets());
517                }
518                (IpAddr::V6(a), IpAddr::V6(b)) => {
519                    out.extend_from_slice(&a.octets());
520                    out.extend_from_slice(&b.octets());
521                }
522                _ => bail!("packet families differ"),
523            };
524            int(&mut out, u32::from(p.source_port));
525            int(&mut out, u32::from(p.destination_port));
526            int(&mut out, p.tcp_flags);
527            int(&mut out, u32::from(p.traffic_class));
528            if ipv6 {
529                4
530            } else {
531                3
532            }
533        }
534        FlowRecord::Ethernet {
535            frame_length,
536            source_mac,
537            destination_mac,
538            ether_type,
539        } => {
540            ensure!(*ether_type <= 65535, "EtherType width");
541            int(&mut out, *frame_length);
542            out.extend_from_slice(&mac(source_mac)?);
543            out.extend_from_slice(&[0, 0]);
544            out.extend_from_slice(&mac(destination_mac)?);
545            out.extend_from_slice(&[0, 0]);
546            int(&mut out, *ether_type);
547            2
548        }
549        FlowRecord::ExtendedSwitch {
550            source_vlan,
551            source_priority,
552            destination_vlan,
553            destination_priority,
554        } => {
555            for v in [*source_vlan, *destination_vlan] {
556                ensure!(v <= 4095 || v == u32::MAX, "VLAN value");
557            }
558            for p in [*source_priority, *destination_priority] {
559                ensure!(p <= 7 || p == u32::MAX, "VLAN priority");
560            }
561            for n in [
562                source_vlan,
563                source_priority,
564                destination_vlan,
565                destination_priority,
566            ] {
567                int(&mut out, *n);
568            }
569            1001
570        }
571    };
572    Ok((tag, out))
573}
574fn synthesized_header(p: &Packet) -> Result<Vec<u8>> {
575    ensure!(
576        p.protocol == 6 || p.protocol == 17,
577        "synthesized headers support TCP/UDP only"
578    );
579    let ipv6 = p.source_ip.is_ipv6();
580    let ip_len = if ipv6 { 40 } else { 20 };
581    let transport_len = if p.protocol == 6 { 20 } else { 8 };
582    ensure!(
583        p.packet_length >= ip_len + transport_len && p.packet_length - ip_len <= 65535,
584        "IP packet length"
585    );
586    let mut b = vec![0; ip_len as usize];
587    match (p.source_ip, p.destination_ip) {
588        (IpAddr::V4(a), IpAddr::V4(c)) => {
589            b[0] = 0x45;
590            b[1] = p.traffic_class;
591            b[2..4].copy_from_slice(&u16::try_from(p.packet_length)?.to_be_bytes());
592            b[8] = 64;
593            b[9] = p.protocol;
594            b[12..16].copy_from_slice(&a.octets());
595            b[16..20].copy_from_slice(&c.octets());
596            let checksum = ipv4_checksum(&b);
597            b[10..12].copy_from_slice(&checksum.to_be_bytes());
598        }
599        (IpAddr::V6(a), IpAddr::V6(c)) => {
600            b[0] = 0x60 | (p.traffic_class >> 4);
601            b[1] = p.traffic_class << 4;
602            b[4..6].copy_from_slice(&((p.packet_length - 40) as u16).to_be_bytes());
603            b[6] = p.protocol;
604            b[7] = 64;
605            b[8..24].copy_from_slice(&a.octets());
606            b[24..40].copy_from_slice(&c.octets());
607        }
608        _ => bail!("synthesized packet family mismatch"),
609    }
610    let at = b.len();
611    b.resize(at + transport_len as usize, 0);
612    b[at..at + 2].copy_from_slice(&p.source_port.to_be_bytes());
613    b[at + 2..at + 4].copy_from_slice(&p.destination_port.to_be_bytes());
614    if p.protocol == 6 {
615        b[at + 12] = 0x50 | ((p.tcp_flags >> 8) as u8);
616        b[at + 13] = p.tcp_flags as u8;
617    } else {
618        b[at + 4..at + 6].copy_from_slice(&((p.packet_length - ip_len) as u16).to_be_bytes());
619    }
620    // These are telemetry header summaries, not captured/live data packets. Payload
621    // and transport checksums are unavailable and are never claimed as validated.
622    Ok(b)
623}
624fn ipv4_checksum(b: &[u8]) -> u16 {
625    let mut sum = b
626        .chunks_exact(2)
627        .map(|w| u32::from(u16::from_be_bytes([w[0], w[1]])))
628        .sum::<u32>();
629    while sum >> 16 != 0 {
630        sum = (sum & 65535) + (sum >> 16);
631    }
632    !(sum as u16)
633}
634pub fn decode(bytes: &[u8]) -> Result<Message> {
635    ensure!(
636        bytes.len() <= MAX_MESSAGE_BYTES,
637        "sFlow message byte limit8192"
638    );
639    let mut r = Reader::new(bytes);
640    ensure!(r.u32()? == 5, "sFlow version5 required");
641    let agent_address = match r.u32()? {
642        1 => IpAddr::V4(Ipv4Addr::from(<[u8; 4]>::try_from(r.take(4)?)?)),
643        2 => IpAddr::V6(Ipv6Addr::from(<[u8; 16]>::try_from(r.take(16)?)?)),
644        _ => bail!("unsupported agent address type"),
645    };
646    let sub_agent_id = r.u32()?;
647    let sequence_number = r.u32()?;
648    let uptime_ms = r.u32()?;
649    let count = r.u32()? as usize;
650    ensure!(count <= MAX_SAMPLES, "sample count limit32");
651    let mut samples = Vec::with_capacity(count);
652    let mut record_count = 0;
653    for _ in 0..count {
654        let tag = r.u32()?;
655        let data = r.opaque(MAX_MESSAGE_BYTES)?;
656        let enterprise = tag >> 12;
657        let format = tag & 4095;
658        if enterprise != 0 || !(1..=4).contains(&format) {
659            ensure!(
660                data.len() <= MAX_OPAQUE_BYTES,
661                "unknown sample byte limit4096"
662            );
663            samples.push(DecodedSample::Unknown {
664                enterprise,
665                format,
666                byte_count: data.len(),
667            });
668            continue;
669        }
670        let flow = format == 1 || format == 3;
671        let expanded = format >= 3;
672        let mut s = Reader::new(data);
673        let sequence_number = s.u32()?;
674        let source = read_source(&mut s, expanded)?;
675        let flow_meta = if flow {
676            Some(FlowMeta {
677                sampling_rate: s.u32()?,
678                sample_pool: s.u32()?,
679                drops: s.u32()?,
680                input: read_interface(&mut s, expanded)?,
681                output: read_interface(&mut s, expanded)?,
682            })
683        } else {
684            None
685        };
686        let n = s.u32()? as usize;
687        ensure!(n <= MAX_RECORDS_PER_SAMPLE, "records/sample limit64");
688        record_count += n;
689        ensure!(record_count <= MAX_RECORDS, "total record limit256");
690        let mut records = Vec::with_capacity(n);
691        for _ in 0..n {
692            let tag = s.u32()?;
693            let b = s.opaque(MAX_OPAQUE_BYTES)?;
694            records.push(decode_record(tag, b, flow)?);
695        }
696        s.done()?;
697        samples.push(if let Some(flow) = flow_meta {
698            DecodedSample::Flow {
699                expanded,
700                sequence_number,
701                source,
702                flow,
703                records,
704            }
705        } else {
706            DecodedSample::Counters {
707                expanded,
708                sequence_number,
709                source,
710                records,
711            }
712        });
713    }
714    r.done()?;
715    Ok(Message {
716        agent_address,
717        sub_agent_id,
718        sequence_number,
719        uptime_ms,
720        samples,
721        record_count,
722        sequence_tracking: SequenceTracking {
723            expected: None,
724            status: "untracked",
725            missing_datagrams: None,
726            uptime_decreased: false,
727        },
728    })
729}
730fn decode_record(tag: u32, bytes: &[u8], flow: bool) -> Result<Record> {
731    let enterprise = tag >> 12;
732    let format = tag & 4095;
733    let mut r = Reader::new(bytes);
734    if enterprise != 0
735        || !if flow {
736            [1, 2, 3, 4, 1001].contains(&format)
737        } else {
738            [1, 2, 5].contains(&format)
739        }
740    {
741        return Ok(Record::Unknown {
742            enterprise,
743            format,
744            byte_count: bytes.len(),
745        });
746    }
747    let record = if flow {
748        match format {
749            1 => {
750                let header_protocol = r.u32()?;
751                let frame_length = r.u32()?;
752                let stripped = r.u32()?;
753                let b = r.opaque(MAX_HEADER_BYTES)?;
754                ensure!(
755                    stripped <= frame_length
756                        && b.len() <= usize::try_from(frame_length - stripped)?,
757                    "sampled header/frame lengths"
758                );
759                let (packet, link_layer, decode_status) = header_summary(header_protocol, b);
760                Record::SampledHeader {
761                    header_protocol,
762                    frame_length,
763                    stripped,
764                    header_length: b.len(),
765                    packet,
766                    link_layer,
767                    decode_status,
768                }
769            }
770            2 => {
771                let frame_length = r.u32()?;
772                let source_mac = mac_text(&r.take(8)?[..6]);
773                let destination_mac = mac_text(&r.take(8)?[..6]);
774                let ether_type = r.u32()?;
775                Record::Ethernet {
776                    frame_length,
777                    source_mac,
778                    destination_mac,
779                    ether_type,
780                }
781            }
782            3 | 4 => {
783                let packet_length = r.u32()?;
784                let protocol = u8::try_from(r.u32()?).context("IP protocol width")?;
785                let ipv6 = format == 4;
786                let mut addr = || -> Result<IpAddr> {
787                    Ok(if ipv6 {
788                        IpAddr::V6(Ipv6Addr::from(<[u8; 16]>::try_from(r.take(16)?)?))
789                    } else {
790                        IpAddr::V4(Ipv4Addr::from(<[u8; 4]>::try_from(r.take(4)?)?))
791                    })
792                };
793                let source_ip = addr()?;
794                let destination_ip = addr()?;
795                let p = Packet {
796                    source_ip,
797                    destination_ip,
798                    packet_length,
799                    protocol,
800                    source_port: u16::try_from(r.u32()?)?,
801                    destination_port: u16::try_from(r.u32()?)?,
802                    tcp_flags: r.u32()?,
803                    traffic_class: u8::try_from(r.u32()?)?,
804                };
805                if ipv6 {
806                    Record::SampledIpv6(p)
807                } else {
808                    Record::SampledIpv4(p)
809                }
810            }
811            1001 => Record::ExtendedSwitch {
812                source_vlan: r.u32()?,
813                source_priority: r.u32()?,
814                destination_vlan: r.u32()?,
815                destination_priority: r.u32()?,
816            },
817            _ => unreachable!(),
818        }
819    } else {
820        match format {
821            1 => Record::InterfaceCounters(InterfaceCounters::decode(&mut r)?),
822            2 => Record::EthernetCounters(EthernetCounters::decode(&mut r)?),
823            5 => Record::VlanCounters(VlanCounters::decode(&mut r)?),
824            _ => unreachable!(),
825        }
826    };
827    r.done()?;
828    Ok(record)
829}
830fn header_summary(
831    protocol: u32,
832    bytes: &[u8],
833) -> (Option<Packet>, Option<LinkLayer>, &'static str) {
834    let mut at = 0;
835    let mut link = None;
836    let ip_protocol = if protocol == 1 {
837        if bytes.len() < 14 {
838            return (None, None, "truncated");
839        }
840        let mut ether_type = u16::from_be_bytes([bytes[12], bytes[13]]);
841        at = 14;
842        let mut vlans = vec![];
843        while ether_type == 0x8100 || ether_type == 0x88a8 {
844            if vlans.len() == 2 {
845                return (None, None, "vlan_depth_limit");
846            }
847            if bytes.len() < at + 4 {
848                return (None, None, "truncated");
849            }
850            vlans.push(u16::from_be_bytes([bytes[at], bytes[at + 1]]) & 4095);
851            ether_type = u16::from_be_bytes([bytes[at + 2], bytes[at + 3]]);
852            at += 4;
853        }
854        link = Some(LinkLayer {
855            source_mac: mac_text(&bytes[6..12]),
856            destination_mac: mac_text(&bytes[..6]),
857            ether_type: u32::from(ether_type),
858            vlans,
859        });
860        match ether_type {
861            0x800 => 11,
862            0x86dd => 12,
863            _ => return (None, link, "unsupported_network_protocol"),
864        }
865    } else {
866        protocol
867    };
868    let b = &bytes[at..];
869    let (mut p, mut offset, non_initial) = match ip_protocol {
870        11 => {
871            if b.len() < 20 {
872                return (None, link, "truncated");
873            }
874            let n = usize::from(b[0] & 15) * 4;
875            if b[0] >> 4 != 4 || n < 20 {
876                return (None, link, "invalid_ip_header");
877            }
878            if b.len() < n {
879                return (None, link, "truncated");
880            }
881            (
882                Packet {
883                    source_ip: IpAddr::V4(Ipv4Addr::new(b[12], b[13], b[14], b[15])),
884                    destination_ip: IpAddr::V4(Ipv4Addr::new(b[16], b[17], b[18], b[19])),
885                    packet_length: u32::from(u16::from_be_bytes([b[2], b[3]])),
886                    protocol: b[9],
887                    source_port: 0,
888                    destination_port: 0,
889                    tcp_flags: 0,
890                    traffic_class: b[1],
891                },
892                n,
893                u16::from_be_bytes([b[6], b[7]]) & 0x1fff != 0,
894            )
895        }
896        12 => {
897            if b.len() < 40 {
898                return (None, link, "truncated");
899            }
900            if b[0] >> 4 != 6 {
901                return (None, link, "invalid_ip_header");
902            }
903            (
904                Packet {
905                    source_ip: IpAddr::V6(Ipv6Addr::from(<[u8; 16]>::try_from(&b[8..24]).unwrap())),
906                    destination_ip: IpAddr::V6(Ipv6Addr::from(
907                        <[u8; 16]>::try_from(&b[24..40]).unwrap(),
908                    )),
909                    packet_length: 40 + u32::from(u16::from_be_bytes([b[4], b[5]])),
910                    protocol: b[6],
911                    source_port: 0,
912                    destination_port: 0,
913                    tcp_flags: 0,
914                    traffic_class: ((b[0] & 15) << 4) | (b[1] >> 4),
915                },
916                40,
917                false,
918            )
919        }
920        _ => return (None, link, "unsupported_header_protocol"),
921    };
922    if non_initial {
923        return (Some(p), link, "non_initial_fragment");
924    }
925    let mut extensions = 0;
926    if ip_protocol == 12 {
927        while [0, 43, 44, 51, 60].contains(&p.protocol) {
928            if extensions == MAX_IPV6_EXTENSIONS {
929                return (Some(p), link, "extension_depth_limit");
930            }
931            extensions += 1;
932            if b.len() < offset + 2 {
933                return (Some(p), link, "truncated");
934            }
935            let n = match p.protocol {
936                44 => 8,
937                51 => (usize::from(b[offset + 1]) + 2) * 4,
938                _ => (usize::from(b[offset + 1]) + 1) * 8,
939            };
940            if b.len() < offset + n {
941                return (Some(p), link, "truncated");
942            }
943            let fragmented = p.protocol == 44
944                && u16::from_be_bytes([b[offset + 2], b[offset + 3]]) & 0xfff8 != 0;
945            p.protocol = b[offset];
946            offset += n;
947            if fragmented {
948                return (Some(p), link, "non_initial_fragment");
949            }
950        }
951    }
952    if p.protocol == 6 || p.protocol == 17 {
953        if b.len() < offset + 4 {
954            return (Some(p), link, "truncated");
955        }
956        p.source_port = u16::from_be_bytes([b[offset], b[offset + 1]]);
957        p.destination_port = u16::from_be_bytes([b[offset + 2], b[offset + 3]]);
958        if p.protocol == 6 {
959            if b.len() < offset + 14 {
960                return (Some(p), link, "truncated");
961            }
962            p.tcp_flags = (u32::from(b[offset + 12] & 1) << 8) | u32::from(b[offset + 13]);
963        }
964    }
965    (Some(p), link, "decoded")
966}
967struct Session {
968    expected: u32,
969    uptime: u32,
970    last_seen: Instant,
971}
972pub struct SequenceCache {
973    sessions: BTreeMap<(SocketAddr, IpAddr, u32), Session>,
974    idle: Duration,
975}
976impl SequenceCache {
977    pub fn new(idle: Duration) -> Self {
978        Self {
979            sessions: BTreeMap::new(),
980            idle,
981        }
982    }
983    pub fn count(&self) -> usize {
984        self.sessions.len()
985    }
986    pub fn expire(&mut self, now: Instant) {
987        self.sessions
988            .retain(|_, s| now.duration_since(s.last_seen) < self.idle);
989    }
990    pub fn ingest(&mut self, peer: SocketAddr, bytes: &[u8], now: Instant) -> Result<Message> {
991        self.expire(now);
992        let mut m = decode(bytes)?;
993        let key = (peer, m.agent_address, m.sub_agent_id);
994        ensure!(
995            self.sessions.contains_key(&key) || self.sessions.len() < MAX_SESSIONS,
996            "sFlow session cap128"
997        );
998        let mut advance = true;
999        if let Some(s) = self.sessions.get(&key) {
1000            let delta = m.sequence_number.wrapping_sub(s.expected);
1001            m.sequence_tracking = SequenceTracking {
1002                expected: Some(s.expected),
1003                status: if delta == 0 {
1004                    "in_order"
1005                } else if delta < 0x80000000 {
1006                    "gap"
1007                } else {
1008                    advance = false;
1009                    "out_of_order_or_duplicate"
1010                },
1011                missing_datagrams: if delta > 0 && delta < 0x80000000 {
1012                    Some(delta)
1013                } else {
1014                    None
1015                },
1016                // A lower uptime can mean a reboot or a late datagram. It does not
1017                // justify regressing sequence state; the idle timeout resets sessions.
1018                uptime_decreased: m.uptime_ms.wrapping_sub(s.uptime) >= 0x80000000,
1019            };
1020        }
1021        if advance {
1022            self.sessions.insert(
1023                key,
1024                Session {
1025                    expected: m.sequence_number.wrapping_add(1),
1026                    uptime: m.uptime_ms,
1027                    last_seen: now,
1028                },
1029            );
1030        } else if let Some(s) = self.sessions.get_mut(&key) {
1031            s.last_seen = now;
1032        }
1033        Ok(m)
1034    }
1035}
netget/server/sflow/codec.rs

netget/server/sflow/
codec.rs
