1use 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;
10pub const DEFAULT_LLM_FALLBACK: bool = false;
11pub const MAX_MESSAGE_BYTES: usize = 8192;
12pub const MAX_SETS: usize = 64;
13pub const MAX_TEMPLATES_PER_SESSION: usize = 32;
14pub const MAX_SESSIONS: usize = 128;
15pub const MAX_CACHED_TEMPLATES: usize = 1024;
16pub const MAX_FIELDS: usize = 32;
17pub const MAX_RECORDS: usize = 256;
18pub const MAX_FIELD_BYTES: usize = 1024;
19pub const DEFAULT_TEMPLATE_TTL: u64 = 600;
20pub const DEFAULT_SESSION_IDLE: u64 = 1800;
21#[derive(Clone, Copy)]
22enum Kind {
23 Counter,
24 Unsigned,
25 Ipv4,
26 Ipv6,
27 Uptime,
28}
29macro_rules! elements {($($variant:ident=>($id:literal,$kind:ident,$width:literal,$name:literal)),* $(,)?)=>{
30 #[derive(Clone,Copy,Debug,Serialize,Deserialize,PartialEq,Eq)]
31 #[serde(rename_all="snake_case")]
32 pub enum Element {$($variant),*}
33 impl Element {
34 pub fn id(self)->u16{match self{$(Self::$variant=>$id),*}}
35 pub fn field_name(self)->&'static str{match self{$(Self::$variant=>$name),*}}
36 fn descriptor(self)->(Kind,u16){match self{$(Self::$variant=>(Kind::$kind,$width)),*}}
37 fn from_id(id:u16)->Option<Self>{match id{$($id=>Some(Self::$variant)),*,_=>None}}
38 fn default_length(self)->u16 {let (kind,width)=self.descriptor();match kind {Kind::Counter=>width.min(4),_=>width}}
39 }
40}}
41elements! {
42 InBytes=>(1,Counter,8,"IN_BYTES"), InPackets=>(2,Counter,8,"IN_PKTS"), Flows=>(3,Counter,8,"FLOWS"),
43 Protocol=>(4,Unsigned,1,"PROTOCOL"), Tos=>(5,Unsigned,1,"TOS"), TcpFlags=>(6,Unsigned,1,"TCP_FLAGS"),
44 SourcePort=>(7,Unsigned,2,"L4_SRC_PORT"), SourceIpv4=>(8,Ipv4,4,"IPV4_SRC_ADDR"), SourceMask=>(9,Unsigned,1,"SRC_MASK"),
45 InputInterface=>(10,Counter,4,"INPUT_SNMP"), DestinationPort=>(11,Unsigned,2,"L4_DST_PORT"), DestinationIpv4=>(12,Ipv4,4,"IPV4_DST_ADDR"),
46 DestinationMask=>(13,Unsigned,1,"DST_MASK"), OutputInterface=>(14,Counter,4,"OUTPUT_SNMP"), NextHopIpv4=>(15,Ipv4,4,"IPV4_NEXT_HOP"),
47 SourceAs=>(16,Counter,4,"SRC_AS"), DestinationAs=>(17,Counter,4,"DST_AS"), BgpNextHopIpv4=>(18,Ipv4,4,"BGP_IPV4_NEXT_HOP"),
48 LastSwitched=>(21,Uptime,4,"LAST_SWITCHED"), FirstSwitched=>(22,Uptime,4,"FIRST_SWITCHED"),
49 OutBytes=>(23,Counter,8,"OUT_BYTES"), OutPackets=>(24,Counter,8,"OUT_PKTS"),
50 SourceIpv6=>(27,Ipv6,16,"IPV6_SRC_ADDR"), DestinationIpv6=>(28,Ipv6,16,"IPV6_DST_ADDR"),
51 SourceIpv6Mask=>(29,Unsigned,1,"IPV6_SRC_MASK"), DestinationIpv6Mask=>(30,Unsigned,1,"IPV6_DST_MASK"),
52 IcmpType=>(32,Unsigned,2,"ICMP_TYPE"), SamplingInterval=>(34,Unsigned,4,"SAMPLING_INTERVAL"), SamplingAlgorithm=>(35,Unsigned,1,"SAMPLING_ALGORITHM"),
53 SourceVlan=>(58,Unsigned,2,"SRC_VLAN"), DestinationVlan=>(59,Unsigned,2,"DST_VLAN"), IpVersion=>(60,Unsigned,1,"IP_PROTOCOL_VERSION"),
54 Direction=>(61,Unsigned,1,"DIRECTION"), NextHopIpv6=>(62,Ipv6,16,"IPV6_NEXT_HOP"), BgpNextHopIpv6=>(63,Ipv6,16,"BGP_IPV6_NEXT_HOP"),
55}
56#[derive(Clone, Copy, Debug, Serialize, Deserialize, PartialEq, Eq)]
57#[serde(rename_all = "snake_case")]
58pub enum Scope {
59 System,
60 Interface,
61 LineCard,
62 Cache,
63 Template,
64}
65impl Scope {
66 fn id(self) -> u16 {
67 match self {
68 Self::System => 1,
69 Self::Interface => 2,
70 Self::LineCard => 3,
71 Self::Cache => 4,
72 Self::Template => 5,
73 }
74 }
75 fn from_id(id: u16) -> Option<Self> {
76 match id {
77 1 => Some(Self::System),
78 2 => Some(Self::Interface),
79 3 => Some(Self::LineCard),
80 4 => Some(Self::Cache),
81 5 => Some(Self::Template),
82 _ => None,
83 }
84 }
85}
86#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)]
87#[serde(
88 tag = "kind",
89 content = "value",
90 rename_all = "snake_case",
91 deny_unknown_fields
92)]
93pub enum FieldValue {
94 Unsigned(u64),
95 Ipv4(Ipv4Addr),
96 Ipv6(Ipv6Addr),
97 UptimeMilliseconds(u32),
98}
99#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)]
100#[serde(deny_unknown_fields)]
101pub struct Field {
102 #[serde(default, skip_serializing_if = "Option::is_none")]
103 pub element: Option<Element>,
104 #[serde(default, skip_serializing_if = "Option::is_none")]
105 pub scope: Option<Scope>,
106 #[serde(default)]
107 pub length: Option<u16>,
108}
109#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)]
110#[serde(deny_unknown_fields)]
111pub struct Template {
112 pub id: u16,
113 #[serde(default)]
114 pub scope_count: u16,
115 pub fields: Vec<Field>,
116}
117#[derive(Clone, Debug, Serialize, Deserialize)]
118#[serde(deny_unknown_fields)]
119pub struct DataSet {
120 pub template_id: u16,
121 pub records: Vec<Vec<FieldValue>>,
122}
123#[derive(Clone, Debug, Serialize, Deserialize)]
124#[serde(deny_unknown_fields)]
125pub struct Batch {
126 pub source_id: u32,
127 #[serde(default)]
128 pub export_time: Option<u32>,
129 #[serde(default)]
130 pub sys_uptime_ms: Option<u32>,
131 pub templates: Vec<Template>,
132 #[serde(default)]
133 pub data_sets: Vec<DataSet>,
134}
135#[derive(Clone, Debug, Serialize, PartialEq, Eq)]
136pub struct WireField {
137 pub field_type: u16,
138 pub scope: Option<Scope>,
139 pub element: Option<Element>,
140 pub length: u16,
141}
142#[derive(Clone, Debug, Serialize, PartialEq, Eq)]
143pub struct WireTemplate {
144 pub id: u16,
145 pub scope_count: u16,
146 pub fields: Vec<WireField>,
147}
148impl Template {
149 pub fn wire(&self) -> Result<WireTemplate> {
150 ensure!(self.id >= 256, "template ID must be256..65535");
151 ensure!(
152 !self.fields.is_empty() && self.fields.len() <= MAX_FIELDS,
153 "template field count1..32"
154 );
155 ensure!(
156 usize::from(self.scope_count) < self.fields.len(),
157 "options templates need scopes followed by at least one option"
158 );
159 let mut fields = Vec::with_capacity(self.fields.len());
160 for (i, f) in self.fields.iter().enumerate() {
161 let scoped = i < usize::from(self.scope_count);
162 ensure!(
163 if scoped {
164 f.scope.is_some() && f.element.is_none()
165 } else {
166 f.element.is_some() && f.scope.is_none()
167 },
168 "fields must declare exactly one scope/element in template order"
169 );
170 let length = f
171 .length
172 .unwrap_or_else(|| f.element.map(Element::default_length).unwrap_or(4));
173 validate_length(f.element, f.scope.is_some(), length)?;
174 fields.push(WireField {
175 field_type: if let Some(e) = f.element {
176 e.id()
177 } else {
178 f.scope.context("scope")?.id()
179 },
180 scope: f.scope,
181 element: f.element,
182 length,
183 });
184 }
185 let t = WireTemplate {
186 id: self.id,
187 scope_count: self.scope_count,
188 fields,
189 };
190 validate_record_size(&t)?;
191 Ok(t)
192 }
193}
194fn validate_length(element: Option<Element>, scoped: bool, length: u16) -> Result<()> {
195 ensure!(
196 length > 0 && usize::from(length) <= MAX_FIELD_BYTES,
197 "fixed field byte bound1..1024; no IPFIX variable-length marker"
198 );
199 if scoped {
200 ensure!(length <= 8, "selected unsigned scope width1..8");
201 } else if let Some(e) = element {
202 let (kind, width) = e.descriptor();
203 match kind {
204 Kind::Counter => ensure!(length <= width, "counter width bound"),
205 _ => ensure!(length == width, "fixed scalar width required"),
206 }
207 }
208 Ok(())
209}
210fn record_size(t: &WireTemplate) -> usize {
211 t.fields.iter().map(|f| usize::from(f.length)).sum()
212}
213fn validate_record_size(t: &WireTemplate) -> Result<()> {
214 ensure!(
215 (4..=MAX_MESSAGE_BYTES).contains(&record_size(t)),
216 "selected record byte size4..8192; shorter records have ambiguous v9 padding"
217 );
218 Ok(())
219}
220fn append(out: &mut Vec<u8>, bytes: &[u8]) -> Result<()> {
221 ensure!(
222 out.len() + bytes.len() <= MAX_MESSAGE_BYTES,
223 "NetFlow v9 message byte bound8192"
224 );
225 out.extend_from_slice(bytes);
226 Ok(())
227}
228fn u16out(out: &mut Vec<u8>, v: u16) -> Result<()> {
229 append(out, &v.to_be_bytes())
230}
231fn set_start(out: &mut Vec<u8>, id: u16) -> Result<usize> {
232 let at = out.len();
233 u16out(out, id)?;
234 u16out(out, 0)?;
235 Ok(at)
236}
237fn set_finish(out: &mut Vec<u8>, at: usize) -> Result<()> {
238 let padding = (4 - (out.len() - at) % 4) % 4;
239 append(out, &[0; 3][..padding])?;
240 let n = (out.len() - at) as u16;
241 out[at + 2..at + 4].copy_from_slice(&n.to_be_bytes());
242 Ok(())
243}
244pub fn encode(
245 batch: &Batch,
246 sequence: u32,
247 export_time: u32,
248 sys_uptime_ms: u32,
249) -> Result<(Vec<u8>, usize)> {
250 ensure!(
251 !batch.templates.is_empty() && batch.templates.len() <= MAX_TEMPLATES_PER_SESSION,
252 "every export must provide1..32 templates"
253 );
254 ensure!(
255 batch.data_sets.len() + batch.templates.len() <= MAX_SETS,
256 "flowset count bound64"
257 );
258 let mut templates = BTreeMap::new();
259 for t in &batch.templates {
260 ensure!(
261 templates.insert(t.id, t.wire()?).is_none(),
262 "duplicate template ID"
263 );
264 }
265 let mut out = Vec::with_capacity(MAX_MESSAGE_BYTES);
266 append(&mut out, &[0, 9, 0, 0])?;
267 for n in [sys_uptime_ms, export_time, sequence, batch.source_id] {
268 append(&mut out, &n.to_be_bytes())?;
269 }
270 for t in templates.values() {
271 let at = set_start(&mut out, if t.scope_count == 0 { 0 } else { 1 })?;
272 u16out(&mut out, t.id)?;
273 if t.scope_count > 0 {
274 u16out(&mut out, t.scope_count * 4)?;
275 u16out(&mut out, (t.fields.len() as u16 - t.scope_count) * 4)?;
276 } else {
277 u16out(&mut out, t.fields.len() as u16)?;
278 }
279 for f in &t.fields {
280 u16out(&mut out, f.field_type)?;
281 u16out(&mut out, f.length)?;
282 }
283 set_finish(&mut out, at)?;
284 }
285 let mut count = 0;
286 for set in &batch.data_sets {
287 let t = templates
288 .get(&set.template_id)
289 .context("every data set needs a template in this batch")?;
290 ensure!(!set.records.is_empty(), "empty data flowset");
291 ensure!(
292 count + set.records.len() <= MAX_RECORDS,
293 "data/options record count bound256"
294 );
295 let at = set_start(&mut out, set.template_id)?;
296 for r in &set.records {
297 ensure!(
298 r.len() == t.fields.len(),
299 "record field count must match template order"
300 );
301 for (f, v) in t.fields.iter().zip(r) {
302 encode_value(&mut out, f, v)?;
303 }
304 }
305 count += set.records.len();
306 set_finish(&mut out, at)?;
307 }
308 let total = (templates.len() + count) as u16;
309 out[2..4].copy_from_slice(&total.to_be_bytes());
310 Ok((out, count))
311}
312fn encode_value(out: &mut Vec<u8>, f: &WireField, v: &FieldValue) -> Result<()> {
313 if f.scope.is_some()
314 || f.element
315 .is_some_and(|e| matches!(e.descriptor().0, Kind::Counter | Kind::Unsigned))
316 {
317 let FieldValue::Unsigned(n) = v else {
318 bail!("unsigned field value required")
319 };
320 ensure!(
321 f.length == 8 || *n < (1u64 << (f.length * 8)),
322 "unsigned value exceeds field width"
323 );
324 append(out, &n.to_be_bytes()[8 - usize::from(f.length)..])?;
325 return Ok(());
326 }
327 let e = f.element.context("outbound unsupported field")?;
328 match (e.descriptor().0, v) {
329 (Kind::Ipv4, FieldValue::Ipv4(ip)) => append(out, &ip.octets())?,
330 (Kind::Ipv6, FieldValue::Ipv6(ip)) => append(out, &ip.octets())?,
331 (Kind::Uptime, FieldValue::UptimeMilliseconds(n)) => append(out, &n.to_be_bytes())?,
332 _ => bail!("value kind does not match {}", e.field_name()),
333 }
334 Ok(())
335}
336struct Reader<'a> {
337 bytes: &'a [u8],
338 at: usize,
339}
340impl<'a> Reader<'a> {
341 fn new(bytes: &'a [u8]) -> Self {
342 Self { bytes, at: 0 }
343 }
344 fn left(&self) -> usize {
345 self.bytes.len() - self.at
346 }
347 fn take(&mut self, n: usize) -> Result<&'a [u8]> {
348 ensure!(n <= self.left(), "truncated NetFlow v9 field/set");
349 let s = &self.bytes[self.at..self.at + n];
350 self.at += n;
351 Ok(s)
352 }
353 fn u16(&mut self) -> Result<u16> {
354 Ok(u16::from_be_bytes(self.take(2)?.try_into()?))
355 }
356 fn u32(&mut self) -> Result<u32> {
357 Ok(u32::from_be_bytes(self.take(4)?.try_into()?))
358 }
359}
360
361#[derive(Clone)]
362struct CachedTemplate {
363 template: WireTemplate,
364 received: Instant,
365}
366#[derive(Clone)]
367struct Session {
368 templates: BTreeMap<u16, CachedTemplate>,
369 expected: Option<u32>,
370 last_seen: Instant,
371 uptime: Option<u32>,
372 export_time: Option<u32>,
373}
374#[derive(Debug, Serialize)]
375pub struct SequenceTracking {
376 pub expected: Option<u32>,
377 pub status: &'static str,
378 pub missing_packets: Option<u32>,
379 pub uptime_decreased: bool,
380 pub cache_reset: Option<&'static str>,
381}
382#[derive(Debug, Serialize)]
383pub struct TemplateChange {
384 pub template: WireTemplate,
385 pub change: &'static str,
386}
387#[derive(Debug, Serialize)]
388pub struct DecodedDataSet {
389 pub template: WireTemplate,
390 pub records: Vec<Vec<Option<FieldValue>>>,
391}
392#[derive(Debug, Serialize)]
393pub struct UnknownDataSet {
394 pub template_id: u16,
395 pub byte_count: usize,
396}
397#[derive(Debug, Serialize)]
398pub struct Message {
399 pub sys_uptime_ms: u32,
400 pub export_time: u32,
401 pub sequence_number: u32,
402 pub source_id: u32,
403 pub header_count: u16,
404 pub known_total_record_count: usize,
405 pub count_status: &'static str,
406 pub sequence_tracking: SequenceTracking,
407 pub template_changes: Vec<TemplateChange>,
408 pub data_sets: Vec<DecodedDataSet>,
409 pub unknown_data_sets: Vec<UnknownDataSet>,
410 pub record_count: usize,
411}
412pub struct TemplateCache {
413 sessions: BTreeMap<(IpAddr, u32), Session>,
414 template_ttl: Duration,
415 session_idle: Duration,
416}
417impl TemplateCache {
418 pub fn new(template_ttl: Duration, session_idle: Duration) -> Self {
419 Self {
420 sessions: BTreeMap::new(),
421 template_ttl,
422 session_idle,
423 }
424 }
425 pub fn counts(&self) -> (usize, usize) {
426 (
427 self.sessions.len(),
428 self.sessions.values().map(|s| s.templates.len()).sum(),
429 )
430 }
431 pub fn expire(&mut self, now: Instant) {
432 self.sessions.retain(|_, s| {
433 s.templates
434 .retain(|_, t| now.duration_since(t.received) < self.template_ttl);
435 now.duration_since(s.last_seen) < self.session_idle
436 });
437 }
438 pub fn ingest(&mut self, peer: SocketAddr, bytes: &[u8], now: Instant) -> Result<Message> {
439 self.expire(now);
440 ensure!(
441 (24..=MAX_MESSAGE_BYTES).contains(&bytes.len()),
442 "NetFlow v9 datagram byte bound24..8192"
443 );
444 let mut r = Reader::new(bytes);
445 ensure!(r.u16()? == 9, "NetFlow v9 version9 required");
446 let header_count = r.u16()?;
447 ensure!(
448 header_count > 0
449 && usize::from(header_count) <= MAX_RECORDS + MAX_TEMPLATES_PER_SESSION,
450 "header total record count bound1..288"
451 );
452 let uptime = r.u32()?;
453 let export_time = r.u32()?;
454 let sequence = r.u32()?;
455 let source_id = r.u32()?;
456 let key = (peer.ip(), source_id);
457 ensure!(
458 self.sessions.contains_key(&key) || self.sessions.len() < MAX_SESSIONS,
459 "source-IP/source-ID session cap128"
460 );
461 let mut candidate = self.sessions.get(&key).cloned().unwrap_or(Session {
462 templates: BTreeMap::new(),
463 expected: None,
464 last_seen: now,
465 uptime: None,
466 export_time: None,
467 });
468 let expected = candidate.expected;
469 let delta = expected.map(|e| sequence.wrapping_sub(e));
470 let advancing = delta.is_none_or(|d| d < 0x8000_0000);
471 let reset = advancing && candidate.export_time.is_some_and(|old| export_time < old);
472 if reset {
473 candidate.templates.clear();
474 }
475 let status = if reset {
476 "clock_regression"
477 } else {
478 match delta {
479 None => "untracked",
480 Some(0) => "in_order",
481 Some(d) if d < 0x8000_0000 => "gap",
482 _ => "out_of_order_or_duplicate",
483 }
484 };
485 let mut message = Message {
486 sys_uptime_ms: uptime,
487 export_time,
488 sequence_number: sequence,
489 source_id,
490 header_count,
491 known_total_record_count: 0,
492 count_status: "validated",
493 sequence_tracking: SequenceTracking {
494 expected,
495 status,
496 missing_packets: delta.filter(|d| *d > 0 && *d < 0x8000_0000),
497 uptime_decreased: candidate.uptime.is_some_and(|old| uptime < old),
498 cache_reset: if reset {
499 Some("clock_regression")
500 } else {
501 None
502 },
503 },
504 template_changes: vec![],
505 data_sets: vec![],
506 unknown_data_sets: vec![],
507 record_count: 0,
508 };
509 let mut sets = 0;
510 let mut template_records = 0;
511 while r.left() > 0 {
512 sets += 1;
513 ensure!(sets <= MAX_SETS, "flowset count bound64");
514 let id = r.u16()?;
515 let length = usize::from(r.u16()?);
516 ensure!(length >= 4, "flowset length includes header");
517 let mut set = Reader::new(r.take(length - 4)?);
518 if id == 0 || id == 1 {
519 let min_header = if id == 0 { 4 } else { 6 };
520 let mut in_set = 0;
521 while set.left() >= min_header {
522 template_records += 1;
523 in_set += 1;
524 ensure!(
525 template_records <= MAX_TEMPLATES_PER_SESSION,
526 "template record count bound32"
527 );
528 let tid = set.u16()?;
529 let first = set.u16()?;
530 let (scope, count) = if id == 1 {
531 let option = set.u16()?;
532 ensure!(
533 first > 0 && option > 0 && first % 4 == 0 && option % 4 == 0,
534 "options scope/option lengths are positive multiples of4 bytes"
535 );
536 (first / 4, usize::from(first / 4) + usize::from(option / 4))
537 } else {
538 (0, usize::from(first))
539 };
540 ensure!(
541 tid >= 256 && (1..=MAX_FIELDS).contains(&count),
542 "template ID/field count bound; v9 has no withdrawal record"
543 );
544 let mut fields = Vec::with_capacity(count);
545 for i in 0..count {
546 let field_type = set.u16()?;
547 let length = set.u16()?;
548 ensure!(field_type != 0, "reserved field type0");
549 let scoped = i < usize::from(scope);
550 let element = if scoped {
551 None
552 } else {
553 Element::from_id(field_type)
554 };
555 let scope_type = if scoped {
556 Scope::from_id(field_type)
557 } else {
558 None
559 };
560 validate_length(element, scope_type.is_some(), length)?;
561 fields.push(WireField {
562 field_type,
563 element,
564 scope: scope_type,
565 length,
566 });
567 }
568 let t = WireTemplate {
569 id: tid,
570 scope_count: scope,
571 fields,
572 };
573 validate_record_size(&t)?;
574 let change = match candidate.templates.get(&tid) {
575 None => Some("new"),
576 Some(old) if old.template != t => Some("replaced"),
577 _ => None,
578 };
579 ensure!(
580 candidate.templates.contains_key(&tid)
581 || candidate.templates.len() < MAX_TEMPLATES_PER_SESSION,
582 "per-source template cap32"
583 );
584 if let Some(change) = change {
585 message.template_changes.push(TemplateChange {
586 template: t.clone(),
587 change,
588 });
589 }
590 candidate.templates.insert(
591 tid,
592 CachedTemplate {
593 template: t,
594 received: now,
595 },
596 );
597 }
598 ensure!(
599 in_set > 0 && set.left() <= 3,
600 "empty template flowset or excessive padding"
601 );
602 } else if id >= 256 {
603 ensure!(set.left() > 0, "empty data flowset");
604 if let Some(cached) = candidate.templates.get(&id) {
605 let t = &cached.template;
606 let size = record_size(t);
607 let mut records = vec![];
608 while set.left() >= size {
609 ensure!(
610 message.record_count < MAX_RECORDS,
611 "data/options record count bound256"
612 );
613 let mut values = Vec::with_capacity(t.fields.len());
614 for f in &t.fields {
615 values.push(decode_value(f, set.take(usize::from(f.length))?)?);
616 }
617 records.push(values);
618 message.record_count += 1;
619 }
620 ensure!(
621 !records.is_empty() && set.left() <= 3,
622 "incomplete data record or excessive padding"
623 );
624 message.data_sets.push(DecodedDataSet {
625 template: t.clone(),
626 records,
627 });
628 } else {
629 message.unknown_data_sets.push(UnknownDataSet {
630 template_id: id,
631 byte_count: set.left(),
632 });
633 }
634 } else {
635 bail!("reserved NetFlow v9 FlowSet ID");
636 }
637 }
638 message.known_total_record_count = template_records + message.record_count;
639 if message.unknown_data_sets.is_empty() {
640 ensure!(
641 message.known_total_record_count == usize::from(header_count),
642 "NetFlow v9 Count must include all template and data/options records"
643 );
644 } else {
645 ensure!(
646 message.known_total_record_count + message.unknown_data_sets.len()
647 <= usize::from(header_count),
648 "each unknown-template flowset needs at least one remaining header record"
649 );
650 message.count_status = "unverifiable_unknown_template";
651 }
652 let old_count = self
653 .sessions
654 .get(&key)
655 .map(|s| s.templates.len())
656 .unwrap_or(0);
657 ensure!(
658 self.counts().1 - old_count + candidate.templates.len() <= MAX_CACHED_TEMPLATES,
659 "global template cap1024"
660 );
661 if advancing {
662 candidate.expected = Some(sequence.wrapping_add(1));
663 candidate.uptime = Some(uptime);
664 candidate.export_time = Some(export_time);
665 }
666 candidate.last_seen = now;
667 self.sessions.insert(key, candidate);
668 Ok(message)
669 }
670}
671fn decode_value(f: &WireField, bytes: &[u8]) -> Result<Option<FieldValue>> {
672 let unsigned = || bytes.iter().fold(0u64, |n, b| (n << 8) | u64::from(*b));
673 if f.scope.is_some() {
674 return Ok(Some(FieldValue::Unsigned(unsigned())));
675 }
676 let Some(e) = f.element else { return Ok(None) };
677 Ok(Some(match e.descriptor().0 {
678 Kind::Counter | Kind::Unsigned => FieldValue::Unsigned(unsigned()),
679 Kind::Ipv4 => FieldValue::Ipv4(Ipv4Addr::from(<[u8; 4]>::try_from(bytes)?)),
680 Kind::Ipv6 => FieldValue::Ipv6(Ipv6Addr::from(<[u8; 16]>::try_from(bytes)?)),
681 Kind::Uptime => FieldValue::UptimeMilliseconds(unsigned() as u32),
682 }))
683}