1use anyhow::{bail, ensure, Context, Result};
3use serde::{Deserialize, Serialize};
4use std::{
5 collections::BTreeMap,
6 net::{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, Debug, PartialEq, Eq)]
22enum Kind {
23 Unsigned,
24 Ipv4,
25 Ipv6,
26 String,
27 Seconds,
28 Milliseconds,
29}
30macro_rules! elements {($($variant:ident=>($id:literal,$kind:ident,$width:literal,$name:literal)),* $(,)?)=>{
31 #[derive(Clone,Copy,Debug,Serialize,Deserialize,PartialEq,Eq)]
32 #[serde(rename_all="snake_case")]
33 pub enum Element {$($variant),*}
34 impl Element {
35 pub fn id(self)->u16{match self{$(Self::$variant=>$id),*}}
36 pub fn iana_name(self)->&'static str{match self{$(Self::$variant=>$name),*}}
37 fn descriptor(self)->(Kind,u16){match self{$(Self::$variant=>(Kind::$kind,$width)),*}}
38 fn from_id(id:u16)->Option<Self>{match id{$($id=>Some(Self::$variant)),*,_=>None}}
39 }
40}}
41elements! {
42 OctetDeltaCount=>(1,Unsigned,8,"octetDeltaCount"),
43 PacketDeltaCount=>(2,Unsigned,8,"packetDeltaCount"),
44 ProtocolIdentifier=>(4,Unsigned,1,"protocolIdentifier"),
45 IpClassOfService=>(5,Unsigned,1,"ipClassOfService"),
46 TcpControlBits=>(6,Unsigned,2,"tcpControlBits"),
47 SourceTransportPort=>(7,Unsigned,2,"sourceTransportPort"),
48 SourceIpv4Address=>(8,Ipv4,4,"sourceIPv4Address"),
49 IngressInterface=>(10,Unsigned,4,"ingressInterface"),
50 DestinationTransportPort=>(11,Unsigned,2,"destinationTransportPort"),
51 DestinationIpv4Address=>(12,Ipv4,4,"destinationIPv4Address"),
52 EgressInterface=>(14,Unsigned,4,"egressInterface"),
53 IpNextHopIpv4Address=>(15,Ipv4,4,"ipNextHopIPv4Address"),
54 BgpSourceAsNumber=>(16,Unsigned,4,"bgpSourceAsNumber"),
55 BgpDestinationAsNumber=>(17,Unsigned,4,"bgpDestinationAsNumber"),
56 SourceIpv6Address=>(27,Ipv6,16,"sourceIPv6Address"),
57 DestinationIpv6Address=>(28,Ipv6,16,"destinationIPv6Address"),
58 SamplingInterval=>(34,Unsigned,4,"samplingInterval"),
59 SamplingAlgorithm=>(35,Unsigned,1,"samplingAlgorithm"),
60 IpVersion=>(60,Unsigned,1,"ipVersion"),
61 InterfaceName=>(82,String,65535,"interfaceName"),
62 InterfaceDescription=>(83,String,65535,"interfaceDescription"),
63 ApplicationName=>(96,String,65535,"applicationName"),
64 ObservationPointId=>(138,Unsigned,8,"observationPointId"),
65 MeteringProcessId=>(143,Unsigned,4,"meteringProcessId"),
66 ExportingProcessId=>(144,Unsigned,4,"exportingProcessId"),
67 TemplateId=>(145,Unsigned,2,"templateId"),
68 ObservationDomainId=>(149,Unsigned,4,"observationDomainId"),
69 FlowStartSeconds=>(150,Seconds,4,"flowStartSeconds"),
70 FlowEndSeconds=>(151,Seconds,4,"flowEndSeconds"),
71 FlowStartMilliseconds=>(152,Milliseconds,8,"flowStartMilliseconds"),
72 FlowEndMilliseconds=>(153,Milliseconds,8,"flowEndMilliseconds"),
73 SystemInitTimeMilliseconds=>(160,Milliseconds,8,"systemInitTimeMilliseconds"),
74 SelectorId=>(302,Unsigned,8,"selectorId"),
75 SelectorAlgorithm=>(304,Unsigned,2,"selectorAlgorithm"),
76 SamplingPacketInterval=>(305,Unsigned,4,"samplingPacketInterval"),
77}
78#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)]
79#[serde(
80 tag = "kind",
81 content = "value",
82 rename_all = "snake_case",
83 deny_unknown_fields
84)]
85pub enum FieldValue {
86 Unsigned(u64),
87 Ipv4(Ipv4Addr),
88 Ipv6(Ipv6Addr),
89 String(String),
90 TimestampSeconds(u32),
91 TimestampMilliseconds(u64),
92}
93#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)]
94#[serde(deny_unknown_fields)]
95pub struct Field {
96 pub element: Element,
97 #[serde(default)]
98 pub length: Option<u16>,
99}
100#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)]
101#[serde(deny_unknown_fields)]
102pub struct Template {
103 pub id: u16,
104 #[serde(default)]
105 pub scope_count: u16,
106 pub fields: Vec<Field>,
107}
108#[derive(Clone, Debug, Serialize, Deserialize)]
109#[serde(deny_unknown_fields)]
110pub struct DataSet {
111 pub template_id: u16,
112 pub records: Vec<Vec<FieldValue>>,
113}
114#[derive(Clone, Debug, Serialize, Deserialize)]
115#[serde(deny_unknown_fields)]
116pub struct Batch {
117 pub observation_domain_id: u32,
118 #[serde(default)]
119 pub export_time: Option<u32>,
120 pub templates: Vec<Template>,
121 #[serde(default)]
122 pub data_sets: Vec<DataSet>,
123}
124#[derive(Clone, Debug, Serialize, PartialEq, Eq)]
125pub struct WireField {
126 pub element_id: u16,
127 pub enterprise: Option<u32>,
128 pub element: Option<Element>,
129 pub length: u16,
130}
131#[derive(Clone, Debug, Serialize, PartialEq, Eq)]
132pub struct WireTemplate {
133 pub id: u16,
134 pub scope_count: u16,
135 pub fields: Vec<WireField>,
136}
137impl Template {
138 pub fn wire(&self) -> Result<WireTemplate> {
139 ensure!(self.id >= 256, "template ID must be256..65535");
140 ensure!(
141 !self.fields.is_empty() && self.fields.len() <= MAX_FIELDS,
142 "template field count1..32"
143 );
144 ensure!(
145 usize::from(self.scope_count) <= self.fields.len(),
146 "scope count exceeds fields"
147 );
148 let fields = self
149 .fields
150 .iter()
151 .map(|f| {
152 let length = f.length.unwrap_or(f.element.descriptor().1);
153 validate_length(Some(f.element), length)?;
154 Ok(WireField {
155 element_id: f.element.id(),
156 enterprise: None,
157 element: Some(f.element),
158 length,
159 })
160 })
161 .collect::<Result<Vec<_>>>()?;
162 let t = WireTemplate {
163 id: self.id,
164 scope_count: self.scope_count,
165 fields,
166 };
167 ensure!(
168 minimum_record_size(&t) <= MAX_MESSAGE_BYTES,
169 "minimum record exceeds message bound"
170 );
171 Ok(t)
172 }
173}
174fn validate_length(element: Option<Element>, length: u16) -> Result<()> {
175 match element.map(Element::descriptor) {
176 Some((Kind::Unsigned, width)) => ensure!(
177 (1..=width).contains(&length),
178 "unsigned field length/reduced-size bound"
179 ),
180 Some((Kind::Ipv4 | Kind::Ipv6 | Kind::Seconds | Kind::Milliseconds, width)) => {
181 ensure!(length == width, "structured scalar cannot use reduced size")
182 }
183 Some((Kind::String, _)) | None => ensure!(
184 length == 65535 || (length > 0 && usize::from(length) <= MAX_FIELD_BYTES),
185 "field length must be1..1024 or65535"
186 ),
187 }
188 Ok(())
189}
190fn minimum_record_size(t: &WireTemplate) -> usize {
191 t.fields
192 .iter()
193 .map(|f| {
194 if f.length == 65535 {
195 1
196 } else {
197 usize::from(f.length)
198 }
199 })
200 .sum()
201}
202fn append(out: &mut Vec<u8>, bytes: &[u8]) -> Result<()> {
203 ensure!(
204 out.len() + bytes.len() <= MAX_MESSAGE_BYTES,
205 "IPFIX message byte bound8192"
206 );
207 out.extend_from_slice(bytes);
208 Ok(())
209}
210fn u16out(out: &mut Vec<u8>, v: u16) -> Result<()> {
211 append(out, &v.to_be_bytes())
212}
213fn set_start(out: &mut Vec<u8>, id: u16) -> Result<usize> {
214 let at = out.len();
215 u16out(out, id)?;
216 u16out(out, 0)?;
217 Ok(at)
218}
219fn set_finish(out: &mut [u8], at: usize) {
220 let n = (out.len() - at) as u16;
221 out[at + 2..at + 4].copy_from_slice(&n.to_be_bytes());
222}
223pub fn encode(batch: &Batch, sequence: u32, export_time: u32) -> Result<(Vec<u8>, usize)> {
224 ensure!(
225 !batch.templates.is_empty() && batch.templates.len() <= MAX_TEMPLATES_PER_SESSION,
226 "every export must provide1..32 templates"
227 );
228 ensure!(
229 batch.data_sets.len() <= MAX_SETS - batch.templates.len(),
230 "set count limit64"
231 );
232 let mut templates = BTreeMap::new();
233 for t in &batch.templates {
234 ensure!(
235 templates.insert(t.id, t.wire()?).is_none(),
236 "duplicate template ID"
237 );
238 }
239 let mut out = Vec::with_capacity(MAX_MESSAGE_BYTES);
240 append(&mut out, &[0, 10, 0, 0])?;
241 append(&mut out, &export_time.to_be_bytes())?;
242 append(&mut out, &sequence.to_be_bytes())?;
243 append(&mut out, &batch.observation_domain_id.to_be_bytes())?;
244 for t in templates.values() {
245 let at = set_start(&mut out, if t.scope_count == 0 { 2 } else { 3 })?;
246 u16out(&mut out, t.id)?;
247 u16out(&mut out, t.fields.len() as u16)?;
248 if t.scope_count > 0 {
249 u16out(&mut out, t.scope_count)?;
250 }
251 for f in &t.fields {
252 u16out(&mut out, f.element_id)?;
253 u16out(&mut out, f.length)?;
254 }
255 set_finish(&mut out, at);
256 }
257 let mut count = 0;
258 for set in &batch.data_sets {
259 let t = templates
260 .get(&set.template_id)
261 .context("every data set needs a template in this batch")?;
262 ensure!(!set.records.is_empty(), "empty data set");
263 ensure!(
264 count + set.records.len() <= MAX_RECORDS,
265 "record count limit256"
266 );
267 let at = set_start(&mut out, set.template_id)?;
268 for r in &set.records {
269 ensure!(
270 r.len() == t.fields.len(),
271 "record field count must match template order"
272 );
273 for (f, v) in t.fields.iter().zip(r) {
274 encode_value(&mut out, f, v)?;
275 }
276 }
277 count += set.records.len();
278 set_finish(&mut out, at);
279 }
280 let n = out.len() as u16;
281 out[2..4].copy_from_slice(&n.to_be_bytes());
282 Ok((out, count))
283}
284fn encode_value(out: &mut Vec<u8>, f: &WireField, v: &FieldValue) -> Result<()> {
285 let element = f.element.context("outbound unsupported element")?;
286 let (kind, _) = element.descriptor();
287 match (kind, v) {
288 (Kind::Unsigned, FieldValue::Unsigned(n)) => {
289 ensure!(
290 f.length == 8 || *n < (1u64 << (f.length * 8)),
291 "unsigned value exceeds field length"
292 );
293 ensure!(
294 element != Element::TcpControlBits || n & 0xf000 == 0,
295 "tcpControlBits data-offset bits must be zero"
296 );
297 append(out, &n.to_be_bytes()[8 - usize::from(f.length)..])?;
298 }
299 (Kind::Ipv4, FieldValue::Ipv4(ip)) => append(out, &ip.octets())?,
300 (Kind::Ipv6, FieldValue::Ipv6(ip)) => append(out, &ip.octets())?,
301 (Kind::Seconds, FieldValue::TimestampSeconds(n)) => append(out, &n.to_be_bytes())?,
302 (Kind::Milliseconds, FieldValue::TimestampMilliseconds(n)) => {
303 append(out, &n.to_be_bytes())?
304 }
305 (Kind::String, FieldValue::String(s)) => {
306 ensure!(s.len() <= MAX_FIELD_BYTES, "UTF-8 field byte bound1024");
307 if f.length == 65535 {
308 if s.len() < 255 {
309 append(out, &[s.len() as u8])?;
310 } else {
311 append(out, &[255])?;
312 u16out(out, s.len() as u16)?;
313 }
314 append(out, s.as_bytes())?;
315 } else {
316 ensure!(
317 s.len() == usize::from(f.length),
318 "fixed string byte count must match declared field length"
319 );
320 append(out, s.as_bytes())?;
321 }
322 }
323 _ => bail!("value kind does not match {}", element.iana_name()),
324 }
325 Ok(())
326}
327struct Reader<'a> {
328 bytes: &'a [u8],
329 at: usize,
330}
331impl<'a> Reader<'a> {
332 fn new(bytes: &'a [u8]) -> Self {
333 Self { bytes, at: 0 }
334 }
335 fn left(&self) -> usize {
336 self.bytes.len() - self.at
337 }
338 fn take(&mut self, n: usize) -> Result<&'a [u8]> {
339 ensure!(n <= self.left(), "truncated IPFIX field/set");
340 let s = &self.bytes[self.at..self.at + n];
341 self.at += n;
342 Ok(s)
343 }
344 fn u16(&mut self) -> Result<u16> {
345 Ok(u16::from_be_bytes(self.take(2)?.try_into()?))
346 }
347 fn u32(&mut self) -> Result<u32> {
348 Ok(u32::from_be_bytes(self.take(4)?.try_into()?))
349 }
350}
351#[derive(Clone)]
352struct CachedTemplate {
353 template: WireTemplate,
354 received: Instant,
355}
356#[derive(Clone)]
357struct Session {
358 templates: BTreeMap<u16, CachedTemplate>,
359 expected: Option<u32>,
360 last_seen: Instant,
361}
362#[derive(Debug, Serialize)]
363pub struct SequenceTracking {
364 pub expected: Option<u32>,
365 pub status: &'static str,
366 pub missing_records: Option<u32>,
367}
368#[derive(Debug, Serialize)]
369pub struct TemplateChange {
370 pub template: WireTemplate,
371 pub change: &'static str,
372}
373#[derive(Debug, Serialize)]
374pub struct DecodedDataSet {
375 pub template: WireTemplate,
376 pub records: Vec<Vec<Option<FieldValue>>>,
377}
378#[derive(Debug, Serialize)]
379pub struct UnknownDataSet {
380 pub template_id: u16,
381 pub byte_count: usize,
382}
383#[derive(Debug, Serialize)]
384pub struct Message {
385 pub export_time: u32,
386 pub sequence_number: u32,
387 pub observation_domain_id: u32,
388 pub sequence_tracking: SequenceTracking,
389 pub template_changes: Vec<TemplateChange>,
390 pub ignored_withdrawals: Vec<u16>,
391 pub data_sets: Vec<DecodedDataSet>,
392 pub unknown_data_sets: Vec<UnknownDataSet>,
393 pub record_count: usize,
394}
395pub struct TemplateCache {
396 sessions: BTreeMap<(SocketAddr, u32), Session>,
397 template_ttl: Duration,
398 session_idle: Duration,
399}
400impl TemplateCache {
401 pub fn new(template_ttl: Duration, session_idle: Duration) -> Self {
402 Self {
403 sessions: BTreeMap::new(),
404 template_ttl,
405 session_idle,
406 }
407 }
408 pub fn counts(&self) -> (usize, usize) {
409 (
410 self.sessions.len(),
411 self.sessions.values().map(|s| s.templates.len()).sum(),
412 )
413 }
414 pub fn expire(&mut self, now: Instant) {
415 self.sessions.retain(|_, s| {
416 s.templates
417 .retain(|_, t| now.duration_since(t.received) < self.template_ttl);
418 now.duration_since(s.last_seen) < self.session_idle
419 });
420 }
421 pub fn ingest(&mut self, peer: SocketAddr, bytes: &[u8], now: Instant) -> Result<Message> {
422 self.expire(now);
423 ensure!(
424 (16..=MAX_MESSAGE_BYTES).contains(&bytes.len()),
425 "IPFIX message byte bound16..8192"
426 );
427 let mut r = Reader::new(bytes);
428 ensure!(r.u16()? == 10, "IPFIX version10 required");
429 ensure!(
430 usize::from(r.u16()?) == bytes.len(),
431 "IPFIX header length must match UDP datagram"
432 );
433 let export_time = r.u32()?;
434 let sequence = r.u32()?;
435 let domain = r.u32()?;
436 let key = (peer, domain);
437 ensure!(
438 self.sessions.contains_key(&key) || self.sessions.len() < MAX_SESSIONS,
439 "IPFIX session/domain cap128"
440 );
441 let mut candidate = self.sessions.get(&key).cloned().unwrap_or(Session {
442 templates: BTreeMap::new(),
443 expected: None,
444 last_seen: now,
445 });
446 let expected = candidate.expected;
447 let delta = expected.map(|e| sequence.wrapping_sub(e));
448 let status = match delta {
449 None => "untracked",
450 Some(0) => "in_order",
451 Some(d) if d < 0x8000_0000 => "gap",
452 Some(_) => "out_of_order_or_duplicate",
453 };
454 let mut message = Message {
455 export_time,
456 sequence_number: sequence,
457 observation_domain_id: domain,
458 sequence_tracking: SequenceTracking {
459 expected,
460 status,
461 missing_records: delta.filter(|d| *d > 0 && *d < 0x8000_0000),
462 },
463 template_changes: vec![],
464 ignored_withdrawals: vec![],
465 data_sets: vec![],
466 unknown_data_sets: vec![],
467 record_count: 0,
468 };
469 let mut sets = 0;
470 let mut template_records = 0;
471 while r.left() > 0 {
472 sets += 1;
473 ensure!(sets <= MAX_SETS, "set count limit64");
474 let id = r.u16()?;
475 let len = usize::from(r.u16()?);
476 ensure!(len >= 4, "set length must include header");
477 let mut set = Reader::new(r.take(len - 4)?);
478 if id == 2 || id == 3 {
479 while set.left() >= 4 {
480 template_records += 1;
481 ensure!(
482 template_records <= MAX_TEMPLATES_PER_SESSION,
483 "template record count limit32"
484 );
485 let tid = set.u16()?;
486 let count = usize::from(set.u16()?);
487 if count == 0 {
488 ensure!(tid >= 256 || tid == id, "invalid UDP withdrawal ID");
489 message.ignored_withdrawals.push(tid);
490 continue;
491 }
492 ensure!(
493 tid >= 256 && count <= MAX_FIELDS,
494 "template ID/field count bound"
495 );
496 let scope = if id == 3 { set.u16()? } else { 0 };
497 ensure!(
498 id != 3 || (scope > 0 && usize::from(scope) <= count),
499 "options scope count must be1..field_count"
500 );
501 let mut fields = Vec::with_capacity(count);
502 for _ in 0..count {
503 let n = set.u16()?;
504 let length = set.u16()?;
505 let enterprise = if n & 0x8000 != 0 {
506 Some(set.u32()?)
507 } else {
508 None
509 };
510 let element_id = n & 0x7fff;
511 ensure!(element_id != 0, "reserved element ID0");
512 let element = if enterprise.is_none() {
513 Element::from_id(element_id)
514 } else {
515 None
516 };
517 validate_length(element, length)?;
518 fields.push(WireField {
519 element_id,
520 enterprise,
521 element,
522 length,
523 });
524 }
525 let t = WireTemplate {
526 id: tid,
527 scope_count: scope,
528 fields,
529 };
530 ensure!(
531 minimum_record_size(&t) <= MAX_MESSAGE_BYTES,
532 "minimum record exceeds message bound"
533 );
534 let change = match candidate.templates.get(&tid) {
535 None => Some("new"),
536 Some(old) if old.template != t => Some("replaced"),
537 _ => None,
538 };
539 ensure!(
540 candidate.templates.contains_key(&tid)
541 || candidate.templates.len() < MAX_TEMPLATES_PER_SESSION,
542 "per-session template cap32"
543 );
544 if let Some(change) = change {
545 message.template_changes.push(TemplateChange {
546 template: t.clone(),
547 change,
548 });
549 }
550 candidate.templates.insert(
551 tid,
552 CachedTemplate {
553 template: t,
554 received: now,
555 },
556 );
557 } } else if id >= 256 {
559 if let Some(cached) = candidate.templates.get(&id) {
560 let t = &cached.template;
561 let min = minimum_record_size(t);
562 let mut records = vec![];
563 while set.left() >= min {
564 ensure!(message.record_count < MAX_RECORDS, "record count limit256");
565 let mut values = Vec::with_capacity(t.fields.len());
566 for f in &t.fields {
567 let n = if f.length == 65535 {
568 let first = set.take(1)?[0];
569 if first == 255 {
570 usize::from(set.u16()?)
571 } else {
572 usize::from(first)
573 }
574 } else {
575 usize::from(f.length)
576 };
577 ensure!(n <= MAX_FIELD_BYTES, "field byte limit1024");
578 let bytes = set.take(n)?;
579 values.push(decode_value(f, bytes)?);
580 }
581 records.push(values);
582 message.record_count += 1;
583 }
584 message.data_sets.push(DecodedDataSet {
585 template: t.clone(),
586 records,
587 });
588 } else {
589 message.unknown_data_sets.push(UnknownDataSet {
590 template_id: id,
591 byte_count: set.left(),
592 });
593 }
594 } else {
595 bail!("reserved IPFIX set ID");
596 }
597 }
598 let old_count = self
599 .sessions
600 .get(&key)
601 .map(|s| s.templates.len())
602 .unwrap_or(0);
603 ensure!(
604 self.counts().1 - old_count + candidate.templates.len() <= MAX_CACHED_TEMPLATES,
605 "global template cap1024"
606 );
607 if !message.unknown_data_sets.is_empty() {
608 candidate.expected = None;
609 } else if delta.is_none_or(|d| d < 0x8000_0000) {
610 candidate.expected = Some(sequence.wrapping_add(message.record_count as u32));
611 }
612 candidate.last_seen = now;
613 self.sessions.insert(key, candidate);
614 Ok(message)
615 }
616}
617fn decode_value(f: &WireField, bytes: &[u8]) -> Result<Option<FieldValue>> {
618 let Some(element) = f.element else {
619 return Ok(None);
620 };
621 let (kind, _) = element.descriptor();
622 let unsigned = || bytes.iter().fold(0u64, |n, b| (n << 8) | u64::from(*b));
623 Ok(Some(match kind {
624 Kind::Unsigned => {
625 let mut n = unsigned();
626 if element == Element::TcpControlBits {
627 n &= 0x0fff;
628 }
629 FieldValue::Unsigned(n)
630 }
631 Kind::Ipv4 => FieldValue::Ipv4(Ipv4Addr::from(<[u8; 4]>::try_from(bytes)?)),
632 Kind::Ipv6 => FieldValue::Ipv6(Ipv6Addr::from(<[u8; 16]>::try_from(bytes)?)),
633 Kind::Seconds => FieldValue::TimestampSeconds(unsigned() as u32),
634 Kind::Milliseconds => FieldValue::TimestampMilliseconds(unsigned()),
635 Kind::String => FieldValue::String(std::str::from_utf8(bytes)?.to_owned()),
636 }))
637}