diff --git a/crates/analytics/src/aggregation.rs b/crates/analytics/src/aggregation.rs index b32cca24..b4fc2392 100644 --- a/crates/analytics/src/aggregation.rs +++ b/crates/analytics/src/aggregation.rs @@ -1,3 +1,4 @@ +// Copyright (C) 2026-present The NetCalyx Authors. // Copyright (C) 2025-present The NetGauze Authors. // // Licensed under the Apache License, Version 2.0 (the "License"); @@ -143,8 +144,9 @@ impl *current_time = (*current_time).max(ts); let window_start = get_window_start(ts); let active_windows = self.active_windows.entry(key).or_default(); - // Aggregates the value in the current window (or create new one if needed) + // Aggregates the value in the current window + // (or create a new one if needed) active_windows .entry(window_start) .or_insert_with(|| AggregatorImpl::init(self.agg_init.clone())) @@ -583,8 +585,9 @@ mod tests { (), ); let (items, expected_results) = get_test_input(); - // Note this doesn't include the final event since the window doesn't close - // without a new event with a timestamp greater than the current time + lateness + // Note this doesn't include the final event since the window + // doesn't close without a new event with a timestamp greater + // than the current time + lateness let expected_on_time: Vec<_> = expected_results[0..expected_results.len() - 1] .iter() .cloned() diff --git a/crates/bgp-pkt/src/codec.rs b/crates/bgp-pkt/src/codec.rs index 5668463f..1fe6146a 100644 --- a/crates/bgp-pkt/src/codec.rs +++ b/crates/bgp-pkt/src/codec.rs @@ -1,3 +1,4 @@ +// Copyright (C) 2026-present The NetCalyx Authors. // Copyright (C) 2023-present The NetGauze Authors. // // Licensed under the Apache License, Version 2.0 (the "License"); @@ -98,7 +99,8 @@ impl Decoder for BgpCodec { if enabled!(Level::TRACE) { trace!(buffer=?buf, length=buf.len(), "decoding buffered message") } - // ASN4 capability is used only when both peers agree on enabling ASN4 + // ASN4 capability is used only when both peers + // agree on enabling ASN4 let asn4 = self.asn4_received.unwrap_or(false) && self.asn4_sent.unwrap_or(false); self.ctx.set_asn4(asn4); let ret = BgpMessage::from_wire(Span::new(buf), &mut self.ctx); diff --git a/crates/bgp-pkt/src/nlri/nlri.rs b/crates/bgp-pkt/src/nlri/nlri.rs index c8cfd7cc..9f1e2a01 100644 --- a/crates/bgp-pkt/src/nlri/nlri.rs +++ b/crates/bgp-pkt/src/nlri/nlri.rs @@ -1,3 +1,4 @@ +// Copyright (C) 2026-present The NetCalyx Authors. // Copyright (C) 2022-present The NetGauze Authors. // // Licensed under the Apache License, Version 2.0 (the "License"); @@ -1252,8 +1253,8 @@ impl Ipv4NlriMplsLabelsAddress { labels: Vec, prefix: Ipv4Net, ) -> Result { - // Total length should not exceed 255, each MPLS Label is 24 bit and account for - // 32 bit IP prefix length + // Total length should not exceed 255, each MPLS Label is 24 bit + // and account for 32 bit IP prefix length if labels.len() * 24 + prefix.prefix_len() as usize > u8::MAX as usize { Err(InvalidIpv4NlriMplsLabelsAddress::InvalidLabelsLength( labels.len(), @@ -1349,8 +1350,8 @@ impl Ipv6NlriMplsLabelsAddress { labels: Vec, prefix: Ipv6Net, ) -> Result { - // Total length should not exceed 255, each MPLS Label is 24 bit and account for - // 32 bit IP prefix length + // Total length should not exceed 255, each MPLS Label is 24 bit + // and account for 32 bit IP prefix length if labels.len() * 24 + prefix.prefix_len() as usize > u8::MAX as usize { Err(InvalidIpv6NlriMplsLabelsAddress::InvalidLabelsLength( labels.len(), diff --git a/crates/bgp-pkt/src/update.rs b/crates/bgp-pkt/src/update.rs index b425e375..748d94c9 100644 --- a/crates/bgp-pkt/src/update.rs +++ b/crates/bgp-pkt/src/update.rs @@ -1,3 +1,4 @@ +// Copyright (C) 2026-present The NetCalyx Authors. // Copyright (C) 2022-present The NetGauze Authors. // // Licensed under the Apache License, Version 2.0 (the "License"); @@ -100,8 +101,8 @@ impl BgpUpdateMessage { if let PathAttributeValue::MpUnreach(unreach) = attr.value() { mp_unreach_count += 1; if mp_unreach_count > 1 { - // Only one MpUnreach is used to indicate End-of-RIB (EoR), more than one - // MpUnreach attribute doesn't define EoR. + // Only one MpUnreach is used to indicate End-of-RIB (EoR), + // more than one MpUnreach attribute doesn't define EoR. return None; } match unreach { @@ -166,8 +167,8 @@ impl BgpUpdateMessage { } } MpUnreach::Unknown { .. } => { - // For unknown address families we assume it's not EoR, as they might have - // different semantics defined. + // For unknown address families we assume it's not EoR, + // as they might have different semantics defined. current = None; } } diff --git a/crates/bgp-pkt/src/wire/deserializer/capabilities.rs b/crates/bgp-pkt/src/wire/deserializer/capabilities.rs index 3248a7db..a670ef32 100644 --- a/crates/bgp-pkt/src/wire/deserializer/capabilities.rs +++ b/crates/bgp-pkt/src/wire/deserializer/capabilities.rs @@ -1,3 +1,4 @@ +// Copyright (C) 2026-present The NetCalyx Authors. // Copyright (C) 2022-present The NetGauze Authors. // // Licensed under the Apache License, Version 2.0 (the "License"); @@ -245,7 +246,8 @@ impl<'a> ReadablePdu<'a, LocatedBgpCapabilityParsingError<'a>> for BgpCapability error: BgpCapabilityParsingError::UndefinedCapabilityCode(UndefinedBgpCapabilityCode(_)), })) => { - // Parse code again, since nom won't advance the buffer on map_res error + // Parse code again, since nom won't advance the buffer + // on map_res error let (buf, code) = be_u8(buf)?; parse_unrecognized_capability(code, buf) } diff --git a/crates/bgp-pkt/src/wire/deserializer/mod.rs b/crates/bgp-pkt/src/wire/deserializer/mod.rs index fe887a04..5759382a 100644 --- a/crates/bgp-pkt/src/wire/deserializer/mod.rs +++ b/crates/bgp-pkt/src/wire/deserializer/mod.rs @@ -1,3 +1,4 @@ +// Copyright (C) 2026-present The NetCalyx Authors. // Copyright (C) 2022-present The NetGauze Authors. // // Licensed under the Apache License, Version 2.0 (the "License"); @@ -241,8 +242,8 @@ impl<'a> ReadablePduWithTwoInputs<'a, u8, Span<'a>, LocatedIpv4PrefixParsingErro prefix_len: u8, prefix_location: Span<'a>, ) -> IResult, Self, LocatedIpv4PrefixParsingError<'a>> { - // The prefix value must fall into the octet boundary, even if the prefix_len - // doesn't. For example, + // The prefix value must fall into the octet boundary, + // even if the prefix_len doesn't. For example, // prefix_len=24 => prefix_size=24 while prefix_len=19 => prefix_size=24 let prefix_size = if prefix_len >= u8::MAX - 7 { u8::MAX @@ -290,8 +291,8 @@ impl<'a> ReadablePduWithTwoInputs<'a, u8, Span<'a>, LocatedIpv6PrefixParsingErro prefix_len: u8, prefix_location: Span<'a>, ) -> IResult, Self, LocatedIpv6PrefixParsingError<'a>> { - // The prefix value must fall into the octet boundary, even if the prefix_len - // doesn't. For example, + // The prefix value must fall into the octet boundary, + // even if the prefix_len doesn't. For example, // prefix_len=24 => prefix_size=24 while prefix_len=19 => prefix_size=24 let prefix_size = if prefix_len >= u8::MAX - 7 { u8::MAX @@ -510,8 +511,8 @@ impl<'a> ReadablePduWithOneInput<'a, &mut BgpParsingContext, LocatedBgpMessagePa } })(buf)?; - // Parse both length and type together, since we need to do input validation on - // the length based on the type of the message + // Parse both length and type together, since we need to do input + // validation on the length based on the type of the message let (buf, (_, message_type, remainder_buf)) = match parse_bgp_message_length_and_type(buf) { Ok(value) => value, Err(err) => return Err(into_located_bgp_message_parsing_error(err)), @@ -580,8 +581,8 @@ impl From for BgpNotificationMessage { BgpNotificationMessage::UpdateMessageError(update_err.into()) } BgpMessageParsingError::BgpNotificationMessageParsingError(_notification) => { - // Notification messages parsing should be ignored and consider a session - // closed. + // Notification messages parsing should be ignored + // and consider a session closed. BgpNotificationMessage::FiniteStateMachineError( FiniteStateMachineError::Unspecific { value: vec![] }, ) diff --git a/crates/bgp-pkt/src/wire/deserializer/nlri/nlri.rs b/crates/bgp-pkt/src/wire/deserializer/nlri/nlri.rs index 9835413d..85273bdb 100644 --- a/crates/bgp-pkt/src/wire/deserializer/nlri/nlri.rs +++ b/crates/bgp-pkt/src/wire/deserializer/nlri/nlri.rs @@ -1,3 +1,4 @@ +// Copyright (C) 2026-present The NetCalyx Authors. // Copyright (C) 2022-present The NetGauze Authors. // // Licensed under the Apache License, Version 2.0 (the "License"); @@ -210,8 +211,8 @@ impl<'a> } else { prefix_len.div_ceil(8) }; - // consuming only the bytes specified by the prefix length field, since MPLS - // stack is read until the last bit is set. + // consuming only the bytes specified by the prefix length field, + // since MPLS stack is read until the last bit is set. let (buf, prefix_buf) = nom::bytes::complete::take(prefix_bytes)(buf)?; let (prefix_buf, label_stack) = parse_mpls_label_stack(prefix_buf, is_unreach, multiple_labels_limit).map_err( @@ -286,8 +287,8 @@ impl<'a> } else { prefix_len.div_ceil(8) }; - // consuming only the bytes specified by the prefix length field, since MPLS - // stack is read until the last bit is set. + // consuming only the bytes specified by the prefix length field, + // since MPLS stack is read until the last bit is set. let (buf, prefix_buf) = nom::bytes::complete::take(prefix_bytes)(buf)?; let (prefix_buf, label_stack) = parse_mpls_label_stack(prefix_buf, is_unreach, multiple_labels_limit).map_err( diff --git a/crates/bgp-pkt/src/wire/deserializer/open.rs b/crates/bgp-pkt/src/wire/deserializer/open.rs index 29dcbb38..5d8154f4 100644 --- a/crates/bgp-pkt/src/wire/deserializer/open.rs +++ b/crates/bgp-pkt/src/wire/deserializer/open.rs @@ -1,3 +1,4 @@ +// Copyright (C) 2026-present The NetCalyx Authors. // Copyright (C) 2022-present The NetGauze Authors. // // Licensed under the Apache License, Version 2.0 (the "License"); @@ -76,10 +77,10 @@ impl<'a> ReadablePduWithOneInput<'a, &mut BgpParsingContext, LocatedBgpOpenMessa let (buf, my_as) = be_u16(buf)?; let begin_buf = buf; let (buf, hold_time) = be_u16(buf)?; - // RFC 4271: If the Hold Time field of the OPEN message is unacceptable, then - // the Error Subcode MUST be set to Unacceptable Hold Time. An implementation - // MUST reject Hold Time values of one or two seconds. An implementation MAY - // reject any proposed Hold Time. + // RFC 4271: If the Hold Time field of the OPEN message is unacceptable, + // then the Error Subcode MUST be set to Unacceptable Hold Time. + // An implementation MUST reject Hold Time values of one or two seconds. + // An implementation MAY reject any proposed Hold Time. if hold_time == 1 || hold_time == 2 { return Err(nom::Err::Error(LocatedBgpOpenMessageParsingError::new( begin_buf, @@ -89,11 +90,11 @@ impl<'a> ReadablePduWithOneInput<'a, &mut BgpParsingContext, LocatedBgpOpenMessa let (buf, bgp_id) = be_u32(buf)?; let begin_buf = buf; let bgp_id = Ipv4Addr::from(bgp_id); - // RFC 4271: If the BGP Identifier field of the OPEN message is syntactically - // incorrect, then the Error Subcode MUST be set to Bad BGP Identifier. - // Syntactic correctness means that the BGP Identifier field represents - // a valid unicast IP host address. NOTE: not all BGP implementation - // check for syntactic correctness + // RFC 4271: If the BGP Identifier field of the OPEN message is + // syntactically incorrect, then the Error Subcode MUST be set to + // Bad BGP Identifier. Syntactic correctness means that the BGP + // Identifier field represents a valid unicast IP host address. + // NOTE: not all BGP implementation check for syntactic correctness if bgp_id.is_broadcast() || bgp_id.is_multicast() || bgp_id.is_unspecified() { return Err(nom::Err::Error(LocatedBgpOpenMessageParsingError::new( begin_buf, @@ -155,11 +156,12 @@ fn parse_capability_param<'a>( nom::Err::Incomplete(needed) => Err(nom::Err::Incomplete(needed))?, nom::Err::Error(err) => { if !ctx.fail_on_capability_error { - // Advance the parser and ignore malformed capability - // RFC 5492 defines that a BGP speaker should ignore capabilities it - // does not understand and not report any error. - // It will only report a notification if the capability is - // understood but not supported by the speaker + // Advance the parser and ignore malformed + // capability. RFC 5492 defines that a BGP speaker + // should ignore capabilities it does not understand + // and not report any error. It will only report + // a notification if the capability is understood + // but not supported by the speaker. let (tmp, _code) = be_u8(capabilities_buf)?; let (tmp, _value) = nom::multi::length_count(be_u8, be_u8)(tmp)?; capabilities_buf = tmp; @@ -172,11 +174,12 @@ fn parse_capability_param<'a>( } nom::Err::Failure(failure) => { if !ctx.fail_on_capability_error { - // Advance the parser and ignore malformed capability - // RFC 5492 defines that a BGP speaker should ignore capabilities it - // does not understand and not report any error. - // It will only report a notification if the capability is - // understood but not supported by the speaker + // Advance the parser and ignore malformed + // capability. RFC 5492 defines that a BGP speaker + // should ignore capabilities it does not understand + // and not report any error. It will only report + // a notification if the capability is understood + // but not supported by the speaker. let (tmp, _code) = be_u8(capabilities_buf)?; let (tmp, _value) = nom::multi::length_count(be_u8, be_u8)(tmp)?; capabilities_buf = tmp; diff --git a/crates/bgp-pkt/src/wire/deserializer/route_refresh.rs b/crates/bgp-pkt/src/wire/deserializer/route_refresh.rs index 5f4e434a..69e464a8 100644 --- a/crates/bgp-pkt/src/wire/deserializer/route_refresh.rs +++ b/crates/bgp-pkt/src/wire/deserializer/route_refresh.rs @@ -1,3 +1,4 @@ +// Copyright (C) 2026-present The NetCalyx Authors. // Copyright (C) 2022-present The NetGauze Authors. // // Licensed under the Apache License, Version 2.0 (the "License"); @@ -69,12 +70,13 @@ impl<'a> ReadablePdu<'a, LocatedBgpRouteRefreshMessageParsingError<'a>> for BgpR impl From for RouteRefreshError { fn from(_value: BgpRouteRefreshMessageParsingError) -> Self { // Mapping all RouteRefresh errors to invalid length - // TODO implement RFC 7313 error handling: If the length, excluding the - // fixed-size message header, of the received ROUTE-REFRESH message with Message - // Subtype 1 and 2 is not 4, then the BGP speaker MUST send a NOTIFICATION - // message with the Error Code of "ROUTE-REFRESH Message Error" and the subcode - // of "Invalid Message Length". The Data field of the NOTIFICATION message MUST - // ontain the complete ROUTE-REFRESH message. + // TODO implement RFC 7313 error handling: If the length, excluding + // the fixed-size message header, of the received ROUTE-REFRESH message + // with Message Subtype 1 and 2 is not 4, then the BGP speaker MUST + // send a NOTIFICATION message with the Error Code of "ROUTE-REFRESH + // Message Error" and the subcode of "Invalid Message Length". + // The Data field of the NOTIFICATION message MUST contain the complete + // ROUTE-REFRESH message. RouteRefreshError::InvalidMessageLength { value: vec![] } } } diff --git a/crates/bgp-pkt/src/wire/deserializer/update.rs b/crates/bgp-pkt/src/wire/deserializer/update.rs index 13a2addd..cb7b837b 100644 --- a/crates/bgp-pkt/src/wire/deserializer/update.rs +++ b/crates/bgp-pkt/src/wire/deserializer/update.rs @@ -1,3 +1,4 @@ +// Copyright (C) 2026-present The NetCalyx Authors. // Copyright (C) 2022-present The NetGauze Authors. // // Licensed under the Apache License, Version 2.0 (the "License"); @@ -77,9 +78,10 @@ fn parse_nlri<'a>( nlri_vec.push(address); } Err(err) => { - // RFC 4271: If a prefix in the NLRI field is semantically incorrect (e.g., an - // unexpected multicast IP address), an error SHOULD be logged locally, and the - // prefix SHOULD be ignored. + // RFC 4271: If a prefix in the NLRI field is semantically + // incorrect (e.g., an unexpected multicast IP address), + // an error SHOULD be logged locally, and the prefix SHOULD + // be ignored. if is_update && ctx.fail_on_non_unicast_update_nlri { ctx.parsing_errors.non_unicast_update_nlri.push(ipv4_net); } @@ -296,11 +298,12 @@ fn handle_path_error<'a>( impl From for UpdateMessageError { fn from(value: BgpUpdateMessageParsingError) -> Self { - // For EoF errors we follow: RFC 4271 Error checking of an UPDATE message begins - // by examining the path attributes. If the Withdrawn Routes Length or - // Total Attribute Length is too large (i.e., if Withdrawn Routes Length - // + Total Attribute Length + 23 exceeds the message Length), then the - // Error Subcode MUST be set to Malformed Attribute List. + // For EoF errors we follow: RFC 4271 Error checking of an UPDATE + // message begins by examining the path attributes. If the + // Withdrawn Routes Length or Total Attribute Length is too large + // (i.e., if Withdrawn Routes Length + Total Attribute Length + 23 + // exceeds the message Length), then the Error Subcode MUST be set + // to Malformed Attribute List. match value { BgpUpdateMessageParsingError::NomError(err) => { if err == nom::error::ErrorKind::Eof { @@ -396,11 +399,11 @@ impl From for UpdateMessageError { UpdateMessageError::InvalidNetworkField { value: vec![] } } BgpUpdateMessageParsingError::InvalidIpv4UnicastNetwork(_) => { - // RFC 4271: If a prefix in the NLRI field is semantically incorrect (e.g., an - // unexpected multicast IP address), an error SHOULD be logged locally, and the - // prefix SHOULD be ignored. - // If parser is configured to be strict and this error triggered, then report - // Unspecific error + // RFC 4271: If a prefix in the NLRI field is semantically + // incorrect (e.g., an unexpected multicast IP address), + // an error SHOULD be logged locally, and the prefix SHOULD + // be ignored. If parser is configured to be strict and this + // error triggered, then report Unspecific error. UpdateMessageError::Unspecific { value: vec![] } } } diff --git a/crates/bgp-pkt/src/wire/serializer/path_attribute/path_attribute.rs b/crates/bgp-pkt/src/wire/serializer/path_attribute/path_attribute.rs index b0323638..33cf2440 100644 --- a/crates/bgp-pkt/src/wire/serializer/path_attribute/path_attribute.rs +++ b/crates/bgp-pkt/src/wire/serializer/path_attribute/path_attribute.rs @@ -1,3 +1,4 @@ +// Copyright (C) 2026-present The NetCalyx Authors. // Copyright (C) 2022-present The NetGauze Authors. // // Licensed under the Apache License, Version 2.0 (the "License"); @@ -177,7 +178,8 @@ pub enum OriginWritingError { } impl WritablePduWithOneInput for Origin { - // One octet length (if extended is not enabled) and second for the origin value + // One octet length (if extended is not enabled) + // and second for the origin value const BASE_LENGTH: usize = 2; fn len(&self, extended_length: bool) -> usize { diff --git a/crates/bgp-pkt/src/wire/tests/path_attribute.rs b/crates/bgp-pkt/src/wire/tests/path_attribute.rs index 859f042c..e9b6660b 100644 --- a/crates/bgp-pkt/src/wire/tests/path_attribute.rs +++ b/crates/bgp-pkt/src/wire/tests/path_attribute.rs @@ -1,3 +1,4 @@ +// Copyright (C) 2026-present The NetCalyx Authors. // Copyright (C) 2022-present The NetGauze Authors. // // Licensed under the Apache License, Version 2.0 (the "License"); @@ -1999,8 +2000,8 @@ fn test_mp_reach_multi_labels_vp_ipv4() -> Result<(), PathAttributeWritingError> &good, ); - // Test with no limit spec, should default to one label and fail since there's - // two labels + // Test with no limit spec, should default to one label + // and fail since there's two labels test_parse_error_with_one_input::< PathAttribute, &mut BgpParsingContext, diff --git a/crates/bgp-pkt/src/wire/tests/pcap_tests.rs b/crates/bgp-pkt/src/wire/tests/pcap_tests.rs index 899934bc..9441f53d 100644 --- a/crates/bgp-pkt/src/wire/tests/pcap_tests.rs +++ b/crates/bgp-pkt/src/wire/tests/pcap_tests.rs @@ -1,3 +1,4 @@ +// Copyright (C) 2026-present The NetCalyx Authors. // Copyright (C) 2024-present The NetGauze Authors. // // Licensed under the Apache License, Version 2.0 (the "License"); @@ -91,8 +92,8 @@ fn test_bgp_pcap(overwrite: bool, pcap_path: PathBuf) { serde_json::to_string(&msg).expect("Couldn't serialize BGP message to json") } Ok(None) => { - // packet is fragmented, need to read the next PDU first before attempting to - // deserialize it + // packet is fragmented, need to read the next PDU first + // before attempting to deserialize it break; } Err(err) => serde_json::to_string(&err) diff --git a/crates/bgp-speaker/src/connection.rs b/crates/bgp-speaker/src/connection.rs index bcb097f2..db6ebcf3 100644 --- a/crates/bgp-speaker/src/connection.rs +++ b/crates/bgp-speaker/src/connection.rs @@ -1,3 +1,4 @@ +// Copyright (C) 2026-present The NetCalyx Authors. // Copyright (C) 2023-present The NetGauze Authors. // // Licensed under the Apache License, Version 2.0 (the "License"); @@ -867,9 +868,9 @@ fn update_treatment(errors: &BgpParsingIgnoredErrors) -> UpdateTreatment { | PathAttributeParsingError::NextHopError(_) | PathAttributeParsingError::MultiExitDiscriminatorError(_) | PathAttributeParsingError::LocalPreferenceError(_) => { - // RFC 7606 "Treat-as-withdraw" MUST be used for the cases that specify a - // session reset and involve any of the attributes ORIGIN, AS_PATH, NEXT_HOP, - // MULTI_EXIT_DISC, or LOCAL_PREF. + // RFC 7606 "Treat-as-withdraw" MUST be used for the cases that + // specify a session reset and involve any of the attributes + // ORIGIN, AS_PATH, NEXT_HOP, MULTI_EXIT_DISC, or LOCAL_PREF. if treatment < UpdateTreatment::TreatAsWithdraw { treatment = UpdateTreatment::TreatAsWithdraw } @@ -884,22 +885,23 @@ fn update_treatment(errors: &BgpParsingIgnoredErrors) -> UpdateTreatment { | PathAttributeParsingError::ExtendedCommunitiesError(_) | PathAttributeParsingError::ExtendedCommunitiesErrorIpv6(_) | PathAttributeParsingError::LargeCommunitiesError(_) => { - // RFC 7606 An UPDATE message with a malformed Community attribute SHALL be - // handled using the approach of "treat-as-withdraw". + // RFC 7606 An UPDATE message with a malformed + // Community attribute SHALL be handled using + // the approach of "treat-as-withdraw". if treatment < UpdateTreatment::TreatAsWithdraw { treatment = UpdateTreatment::TreatAsWithdraw } } PathAttributeParsingError::OriginatorError(_) => { - // RFC 7606 If malformed, the UPDATE message SHALL be handled using the - // approach of "treat-as-withdraw". + // RFC 7606 If malformed, the UPDATE message SHALL be handled + // using the approach of "treat-as-withdraw". if treatment < UpdateTreatment::TreatAsWithdraw { treatment = UpdateTreatment::TreatAsWithdraw } } PathAttributeParsingError::ClusterListError(_) => { - // RFC 7606 If malformed, the UPDATE message SHALL be handled using the - // approach of "treat-as-withdraw". + // RFC 7606 If malformed, the UPDATE message SHALL be handled + // using the approach of "treat-as-withdraw". if treatment < UpdateTreatment::TreatAsWithdraw { treatment = UpdateTreatment::TreatAsWithdraw } @@ -914,8 +916,8 @@ fn update_treatment(errors: &BgpParsingIgnoredErrors) -> UpdateTreatment { } MpReachParsingError::UndefinedAddressFamily(_) | MpReachParsingError::UndefinedSubsequentAddressFamily(_) => { - // AFI/SAFI is not supported, this would've been blocked from open message - // in the first place + // AFI/SAFI is not supported, this would've been blocked + // from open message in the first place if treatment < UpdateTreatment::SessionReset { treatment = UpdateTreatment::SessionReset } @@ -1049,8 +1051,8 @@ fn update_treatment(errors: &BgpParsingIgnoredErrors) -> UpdateTreatment { } MpUnreachParsingError::UndefinedAddressFamily(_) | MpUnreachParsingError::UndefinedSubsequentAddressFamily(_) => { - // AFI/SAFI is not supported, this would've been blocked from open message - // in the first place + // AFI/SAFI is not supported, this would've been blocked + // from open message in the first place if treatment < UpdateTreatment::SessionReset { treatment = UpdateTreatment::SessionReset } @@ -1170,11 +1172,12 @@ fn update_treatment(errors: &BgpParsingIgnoredErrors) -> UpdateTreatment { // Keep treatment as is } PathAttributeParsingError::InvalidPathAttribute(err, _) => { - // RFC 7606: If the value of either the Optional or Transitive bits in the - // Attribute Flags is in conflict with their specified values, then the - // attribute MUST be treated as malformed and the "treat-as-withdraw" approach - // used, unless the specification for the attribute mandates different handling - // for incorrect Attribute Flags. + // RFC 7606: If the value of either the Optional or Transitive + // bits in the Attribute Flags is in conflict with their + // specified values, then the attribute MUST be treated + // as malformed and the "treat-as-withdraw" approach used, + // unless the specification for the attribute mandates + // different handling for incorrect Attribute Flags. match err { InvalidPathAttribute::InvalidOptionalFlagValue(_) | InvalidPathAttribute::InvalidTransitiveFlagValue(_) => { @@ -1218,10 +1221,10 @@ fn handle_open_message( }), ); } - // TODO: check BGP ID according to RFC4271: If the BGP Identifier field of the - // OPEN message is syntactically incorrect, then the Error Subcode MUST be set - // to Bad BGP Identifier. Syntactic correctness means that the BGP Identifier - // field represents a valid unicast IP host address. + // TODO: check BGP ID according to RFC4271: If the BGP Identifier field + // of the OPEN message is syntactically incorrect, then the Error Subcode + // MUST be set to Bad BGP Identifier. Syntactic correctness means that + // the BGP Identifier field represents a valid unicast IP host address. if delay_timer_running { ( @@ -1237,9 +1240,9 @@ fn handle_update_message( update: BgpUpdateMessage, parsing_errors: BgpParsingIgnoredErrors, ) -> Option> { - // RFC 7606 If any of the well-known mandatory attributes are not present in an - // UPDATE message, then "treat-as-withdraw" MUST be used. (Note that [RFC4760] - // reclassifies NEXT_HOP as what is effectively discretionary.) + // RFC 7606: If any of the well-known mandatory attributes are not present + // in an UPDATE message, then "treat-as-withdraw" MUST be used. (Note that + // [RFC4760] reclassifies NEXT_HOP as what is effectively discretionary.) let end_of_rib = update.end_of_rib(); let mut has_origin = false; let mut has_asn_path = false; @@ -1284,8 +1287,9 @@ fn handle_update_message( && !has_next_hop && !update.nlri().is_empty() { - // RFC7606: RFC4760 reclassifies NEXT_HOP as what is effectively discretionary. - // Complain if BGP-MP is not used and there are reachable NLRI announced. + // RFC7606: RFC4760 reclassifies NEXT_HOP as what is effectively + // discretionary. Complain if BGP-MP is not used and there are + // reachable NLRI announced. return Some(ConnectionEvent::UpdateMsgErr( UpdateMessageError::MissingWellKnownAttribute { value: vec![PathAttributeType::NextHop as u8], @@ -1293,9 +1297,10 @@ fn handle_update_message( )); } if bgp_mp_reach_count > 1 || bgp_mp_unreach_count > 1 { - // RFC7606: If the MP_REACH_NLRI attribute or the MP_UNREACH_NLRI [RFC4760] - // attribute appears more than once in the UPDATE message, then a NOTIFICATION - // message MUST be sent with the Error Subcode "Malformed Attribute List". + // RFC7606: If the MP_REACH_NLRI attribute or the MP_UNREACH_NLRI + // [RFC4760] attribute appears more than once in the UPDATE message, + // then a NOTIFICATION message MUST be sent with the Error Subcode + // "Malformed Attribute List". return Some(ConnectionEvent::UpdateMsgErr( UpdateMessageError::MalformedAttributeList { value: vec![] }, )); diff --git a/crates/bgp-speaker/src/listener.rs b/crates/bgp-speaker/src/listener.rs index 199960c2..17acac2a 100644 --- a/crates/bgp-speaker/src/listener.rs +++ b/crates/bgp-speaker/src/listener.rs @@ -1,3 +1,4 @@ +// Copyright (C) 2026-present The NetCalyx Authors. // Copyright (C) 2023-present The NetGauze Authors. // // Licensed under the Apache License, Version 2.0 (the "License"); @@ -109,7 +110,8 @@ impl BgpListener { if !self.allow_dynamic_peers { info!("No peer configured for: {peer_addr}"); } else { - // TODO: rewrite for more clear logic and dynamic peer handling factory + // TODO: rewrite for more clear logic + // and dynamic peer handling factory if let Ok((mut rx, mut peer_handle)) = peer_supervisor.dynamic_peer(peer_key, peer_addr, TcpActiveConnect) { diff --git a/crates/bgp-speaker/src/peer.rs b/crates/bgp-speaker/src/peer.rs index bf5286dd..59e3e68f 100644 --- a/crates/bgp-speaker/src/peer.rs +++ b/crates/bgp-speaker/src/peer.rs @@ -1,3 +1,4 @@ +// Copyright (C) 2026-present The NetCalyx Authors. // Copyright (C) 2023-present The NetGauze Authors. // // Licensed under the Apache License, Version 2.0 (the "License"); @@ -162,8 +163,8 @@ impl< } for cap in &self.peer_capabilities { - // Check that the capability has not been added before and not in the reject - // list + // Check that the capability has not been added before + // and not in the reject list if !self.capabilities.contains(cap) && !self.reject_capabilities.contains(cap) { capabilities.push(cap.clone()); } @@ -172,8 +173,8 @@ impl< let params = if capabilities.is_empty() { vec![] } else { - // TODO check for param size and spread capabilities across multiple params or - // use extended params RFC 9072 + // TODO check for param size and spread capabilities across + // multiple params or use extended params RFC 9072 vec![BgpOpenMessageParameter::Capabilities(capabilities)] }; @@ -709,7 +710,8 @@ impl< }); let codec = D::new(self); let mut framed = Framed::new(tcp_stream, codec); - // Error is ignored since it's optional to send a notification message + // Error is ignored since it's optional to send + // a notification message let _ = framed.send(BgpMessage::Notification(notif)).await; let _ = framed.close().await; } @@ -735,8 +737,8 @@ impl< if let Some(tracked) = self.tracked_connection.as_mut() && let Err(err) = tracked.send(msg.clone()).await { - // Errors writing to a tracked connection are ignored and we assume that the - // connection is not good anymore. + // Errors writing to a tracked connection are ignored + // and we assume that the connection is not good anymore. info!( "[{}][{}] Error writing to tracked connection at state {} : {err:?}", self.peer_key, @@ -935,7 +937,8 @@ impl< | ConnectionEvent::TcpConnectionConfirmed(_) => { self.connect_retry_timer.take(); if conn.open_delay_timer().is_none() { - // Only allowed to stay in this state if open delay timer is running + // Only allowed to stay in this state + // if open delay timer is running return Err(FsmStateError::InvalidConnectionStateTransition( event.clone().into(), self.fsm_state, @@ -1401,10 +1404,10 @@ impl< Some((tracked_peer_bgp_id, tracked_created)), ) = (main_info, tracked_info) { - // This is not part of the BGP Spec, currently it's not defined if the BGP - // Peer ID signaled in main and tracked connections are different. - // We take the one in the main connection as the reference one and close the - // tracked connection. + // This is not part of the BGP Spec, currently it's not defined + // if the BGP Peer ID signaled in main and tracked connections + // are different. We take the one in the main connection as the + // reference one and close the tracked connection. if tracked_peer_bgp_id != main_peer_bgp_id { return Some(CollisionCheckRet::InvalidTrackedBgpId(tracked_peer_bgp_id)); } @@ -1430,9 +1433,9 @@ impl< mut connection: Option<&mut Connection>, mut tracked_connection: Option<&mut Connection>, ) -> ConnectionNextEvent { - // Looping to till one event is produced. Note this is because we ignore tracked - // connection events and we wait for either main connection event or a - // collision detection event. + // Looping until one event is produced. Note this is because + // we ignore tracked connection events and we wait for either + // a main connection event or a collision detection event. loop { let event = tokio::select! { event = Self::get_connection_event(&mut connection) => { diff --git a/crates/bgp-speaker/src/tests/connection.rs b/crates/bgp-speaker/src/tests/connection.rs index bbef0c44..2c409300 100644 --- a/crates/bgp-speaker/src/tests/connection.rs +++ b/crates/bgp-speaker/src/tests/connection.rs @@ -418,8 +418,8 @@ async fn test_open_confirm_hold_timer_expires() -> io::Result<()> { let event = connection.handle_event(&mut policy, event).await; assert_eq!(event, Ok(ConnectionEvent::KeepAliveTimerExpires)); - // Receive and handle HoldTimer expire after multiple Keep Alive messages sent - // without response + // Receive and handle HoldTimer expire after multiple Keep Alive messages + // sent without response let event = tokio::time::timeout(Duration::from_secs(hold_time_seconds), connection.next()).await; assert_eq!(event, Ok(Some(ConnectionEvent::HoldTimerExpires))); diff --git a/crates/bgp-speaker/src/tests/peer.rs b/crates/bgp-speaker/src/tests/peer.rs index a74af18e..90f73a39 100644 --- a/crates/bgp-speaker/src/tests/peer.rs +++ b/crates/bgp-speaker/src/tests/peer.rs @@ -985,11 +985,11 @@ async fn test_active_manual_start() { let event = peer.run().await.unwrap(); assert_eq!(event, BgpEvent::ManualStartWithPassiveTcp); assert_eq!(peer.fsm_state(), FsmState::Active); - // // Start should be ignored + // Start should be ignored peer.add_admin_event(PeerAdminEvents::ManualStart); let event = tokio::time::timeout(Duration::from_millis(1), peer.run()).await; - // since manual start is ignored, and no connection is added, no more new events - // should be returned by run + // since manual start is ignored, and no connection is added, + // no more new events should be returned by run assert!(event.is_err()); assert!(peer.waiting_admin_events().is_empty()); assert_eq!(peer.fsm_state(), FsmState::Active); @@ -1016,11 +1016,11 @@ async fn test_active_automatic_start() { let event = peer.run().await.unwrap(); assert_eq!(event, BgpEvent::ManualStartWithPassiveTcp); assert_eq!(peer.fsm_state(), FsmState::Active); - // // Start should be ignored + // Start should be ignored peer.add_admin_event(PeerAdminEvents::AutomaticStart); let event = tokio::time::timeout(Duration::from_millis(1), peer.run()).await; - // since start is ignored, and no connection is added, no more new events should - // be returned by run + // since start is ignored, and no connection is added, + // no more new events should be returned by run assert!(event.is_err()); assert!(peer.waiting_admin_events().is_empty()); assert_eq!(peer.fsm_state(), FsmState::Active); diff --git a/crates/bmp-pkt/src/codec.rs b/crates/bmp-pkt/src/codec.rs index 77d1efcf..8a5c3484 100644 --- a/crates/bmp-pkt/src/codec.rs +++ b/crates/bmp-pkt/src/codec.rs @@ -1,3 +1,4 @@ +// Copyright (C) 2026-present The NetCalyx Authors. // Copyright (C) 2022-present The NetGauze Authors. // // Licensed under the Apache License, Version 2.0 (the "License"); @@ -129,21 +130,23 @@ impl BmpParsingContext { // Add Key for the router announcing BMP to the collector let peer_key = PeerKey::from_peer_header(peer_up.peer_header()); let bgp_ctx = ctx.entry(peer_key).or_default(); - // According to [RFC 9069 Section 6.1.1](https://datatracker.ietf.org/doc/html/rfc9069#name-multiple-loc-rib-peers) - // In some implementations, it might be required to have more than one emulated - // peer for Loc-RIB to convey different address families for the - // same Loc-RIB. In this case, the peer distinguisher and BGP ID - // should be the same since they represent the same Loc-RIB - // instance. Each emulated peer instance MUST send a Peer Up with - // the OPEN message indicating the address family capabilities. - // A BMP receiver MUST process these capabilities to know which peer belongs to - // which address family. + // According to + // [RFC 9069 Section 6.1.1](https://datatracker.ietf.org/doc/html/rfc9069#name-multiple-loc-rib-peers) + // In some implementations, it might be required to have more + // than one emulated peer for Loc-RIB to convey different address + // families for the same Loc-RIB. In this case, the peer + // distinguisher and BGP ID should be the same since they represent + // the same Loc-RIB instance. Each emulated peer instance MUST send + // a Peer Up with the OPEN message indicating the address family + // capabilities. A BMP receiver MUST process these capabilities + // to know which peer belongs to which address family. if !matches!(peer_key.peer_type(), BmpPeerType::LocRibInstancePeer { .. }) { bgp_ctx.add_path_mut().clear(); bgp_ctx.multiple_labels_mut().clear(); } // Determine if we need to track Adj-RIB-Out based on Peer Type, - // which is useful to select ADD-Path behavior for either sending or receive + // which is useful to select ADD-Path behavior for either sending + // or receiving let adj_rib_out = match peer_up.peer_header().peer_type() { BmpPeerType::GlobalInstancePeer { adj_rib_out, .. } | BmpPeerType::RdInstancePeer { adj_rib_out, .. } @@ -167,8 +170,8 @@ impl BmpParsingContext { ); // Add a key for the BGP Peer of the first router - // In Loc-Rib the bgp open message is duplicated, no need to go through it - // again. + // In Loc-Rib the bgp open message is duplicated, + // no need to go through it again. if !matches!(peer_key.peer_type(), BmpPeerType::LocRibInstancePeer { .. }) { let peer_key = PeerKey::new( peer_up.peer_header().address(), @@ -179,7 +182,8 @@ impl BmpParsingContext { ); let bgp_ctx = ctx.entry(peer_key).or_default(); // Determine if we need to track Adj-RIB-Out based on Peer Type, - // which is useful to select ADD-Path behavior for either sending or receive + // which is useful to select ADD-Path behavior for either + // sending or receiving let adj_rib_out = match peer_up.peer_header().peer_type() { BmpPeerType::GlobalInstancePeer { adj_rib_out, .. } | BmpPeerType::RdInstancePeer { adj_rib_out, .. } @@ -286,10 +290,11 @@ impl Decoder for BmpCodec { BmpCodecDecoderError::BmpMessageParsingError(error.error().clone()) } }; - // Make sure we advance the buffer far enough, so we don't get stuck on an - // error value. - // Unfortunately, BMP doesn't have synchronization values like in BGP - // to understand we are in a new message. + // Make sure we advance the buffer far enough, + // so we don't get stuck on an error value. + // Unfortunately, BMP doesn't have synchronization + // values like in BGP to understand we are in a new + // message. buf.advance(if length < 5 { 5 } else { length }); return Err(err); } @@ -526,14 +531,14 @@ mod tests { buf.extend_from_slice(&up1_wire); buf.extend_from_slice(&up2_wire); - // check after each decoded BGP open the ADD Path ctx is including the new - // address family + // check after each decoded BGP open if the ADD Path ctx + // is including the new address family let _ = codec .decode(&mut buf) .expect("decode up1_wire failed") .expect("no message decoded from up1"); - // Only IPv4 add path is added to the BGP decoding context for the first peer up - // message + // Only IPv4 add path is added to the BGP decoding context + // for the first peer up message let add_path1 = codec .ctx .get_peer(&peer_key) @@ -547,8 +552,8 @@ mod tests { .decode(&mut buf) .expect("decode up1_wire failed") .expect("no message decoded from up2"); - // IPv6 add path is added to the BGP decoding context without deleting the add - // path for IPv4 + // IPv6 add path is added to the BGP decoding context + // without deleting the add path for IPv4 let add_path2 = codec .ctx .get_peer(&peer_key) diff --git a/crates/bmp-pkt/src/wire/deserializer/v4.rs b/crates/bmp-pkt/src/wire/deserializer/v4.rs index 65377ad8..6270f588 100644 --- a/crates/bmp-pkt/src/wire/deserializer/v4.rs +++ b/crates/bmp-pkt/src/wire/deserializer/v4.rs @@ -1,3 +1,4 @@ +// Copyright (C) 2026-present The NetCalyx Authors. // Copyright (C) 2022-present The NetGauze Authors. // // Licensed under the Apache License, Version 2.0 (the "License"); @@ -181,7 +182,8 @@ impl<'a> bgp_ctx.set_asn4(peer_header.is_asn4()); // Determine if we need to track Adj-RIB-Out based on Peer Type, - // which is useful to select ADD-Path behavior for either sending or receive + // which is useful to select ADD-Path behavior for either sending + // or receiving let adj_rib_out = match peer_header.peer_type() { BmpPeerType::GlobalInstancePeer { adj_rib_out, .. } | BmpPeerType::RdInstancePeer { adj_rib_out, .. } @@ -190,8 +192,8 @@ impl<'a> }; // Context represents what we learnt from the BGP Open - // We do not want to alter it permanently based on TLVs that are punctual in the - // messages + // We do not want to alter it permanently based on TLVs + // that are punctual in the messages let mut ctx_clone = bgp_ctx.clone(); // Can't use parse_till_empty_into_with_one_input_located because @@ -202,9 +204,9 @@ impl<'a> let mut tlvs = Vec::new(); let mut bgp_pdu = None; while !buf.is_empty() { - // Peek the TLV Type, if we have a BGP PDU we keep it for later and we'll decode - // it when we've decoded all the Stateless Parsing TLVs on which - // the PDU decoding depends + // Peek the TLV Type, if we have a BGP PDU we keep it for later + // and we'll decode it when we've decoded all the Stateless + // Parsing TLVs on which the PDU decoding depends match nom::combinator::peek(be_u16)(buf)? { (_, tlv_type) if tlv_type == v4::RouteMonitoringTlvType::BgpUpdatePdu as u16 => @@ -302,8 +304,8 @@ impl<'a> ctx: &mut BgpParsingContext, adj_rib_out: bool, ) -> IResult, Self, LocatedRouteMonitoringTlvParsingError<'a>> { - // Can't use read_tlv_header_t16_l16 because Index is in the middle of the - // header and not counted in Length + // Can't use read_tlv_header_t16_l16 because Index is in the middle + // of the header and not counted in Length let (span, tlv_type) = be_u16(buf)?; let input = buf; let (span, tlv_length) = be_u16(span)?; diff --git a/crates/bmp-pkt/src/wire/tests/pcap_tests.rs b/crates/bmp-pkt/src/wire/tests/pcap_tests.rs index 3d864c1c..79a9bdf9 100644 --- a/crates/bmp-pkt/src/wire/tests/pcap_tests.rs +++ b/crates/bmp-pkt/src/wire/tests/pcap_tests.rs @@ -1,3 +1,4 @@ +// Copyright (C) 2026-present The NetCalyx Authors. // Copyright (C) 2024-present The NetGauze Authors. // // Licensed under the Apache License, Version 2.0 (the "License"); @@ -93,8 +94,8 @@ fn test_bmp_pcap(overwrite: bool, pcap_path: PathBuf) { serde_json::to_string(&msg).expect("Couldn't serialize BMP message to json") } Ok(None) => { - // packet is fragmented, need to read the next PDU first before attempting to - // deserialize it + // packet is fragmented, need to read the next PDU first + // before attempting to deserialize it break; } Err(err) => serde_json::to_string(&err) diff --git a/crates/bmp-service/examples/bmp-actor-example.rs b/crates/bmp-service/examples/bmp-actor-example.rs index a17ddcb9..50682ed0 100644 --- a/crates/bmp-service/examples/bmp-actor-example.rs +++ b/crates/bmp-service/examples/bmp-actor-example.rs @@ -1,3 +1,4 @@ +// Copyright (C) 2026-present The NetCalyx Authors. // Copyright (C) 2026-present The NetGauze Authors. // // Licensed under the Apache License, Version 2.0 (the "License"); @@ -130,11 +131,12 @@ pub fn main() -> Result<(), Box> // Spawn a task to print received BMP messages tokio::spawn(async move { while let Ok(pkt) = pkt_rx.recv().await { - // pkt: Arc where BmpRequest = (AddrInfo, BmpMessage) + // pkt: Arc where BmpRequest = (AddrInfo, + // BmpMessage) let (addrinfo, bmp_msg) = &*pkt; - // try to produce a JSON representation of the BMP message, fall back to debug - // if serialization fails + // try to produce a JSON representation of the BMP message, + // fall back to debug if serialization fails let json_msg = match serde_json::to_string(&bmp_msg) { Ok(s) => s, Err(e) => { @@ -143,8 +145,8 @@ pub fn main() -> Result<(), Box> } }; - // use tracing structured fields and print AddrInfo inside brackets plus the - // JSON message + // use tracing structured fields and print AddrInfo inside + // brackets plus the JSON message tracing::info!( local_addr = %addrinfo.local_socket(), peer_addr = %addrinfo.remote_socket(), diff --git a/crates/bmp-service/examples/bmp-supervisor-example.rs b/crates/bmp-service/examples/bmp-supervisor-example.rs index a84269a7..32679b85 100644 --- a/crates/bmp-service/examples/bmp-supervisor-example.rs +++ b/crates/bmp-service/examples/bmp-supervisor-example.rs @@ -1,3 +1,4 @@ +// Copyright (C) 2026-present The NetCalyx Authors. // Copyright (C) 2026-present The NetGauze Authors. // // Licensed under the Apache License, Version 2.0 (the "License"); @@ -95,11 +96,12 @@ pub fn main() -> Result<(), Box> // Spawn a task to print received BMP messages tokio::spawn(async move { while let Ok(pkt) = pkt_rx.recv().await { - // pkt: Arc where BmpRequest = (AddrInfo, BmpMessage) + // pkt: Arc where BmpRequest = (AddrInfo, + // BmpMessage) let (addrinfo, bmp_msg) = &*pkt; - // try to produce a JSON representation of the BMP message, fall back to debug - // if serialization fails + // try to produce a JSON representation of the BMP message, + // fall back to debug if serialization fails let json_msg = match serde_json::to_string(&bmp_msg) { Ok(s) => s, Err(e) => { @@ -108,8 +110,8 @@ pub fn main() -> Result<(), Box> } }; - // use tracing structured fields and print AddrInfo inside brackets plus the - // JSON message + // use tracing structured fields and print AddrInfo inside + // brackets plus the JSON message tracing::info!( local_addr = %addrinfo.local_socket(), peer_addr = %addrinfo.remote_socket(), diff --git a/crates/bmp-service/examples/print-bmp.rs b/crates/bmp-service/examples/print-bmp.rs index 6df38b44..e0bf27d8 100644 --- a/crates/bmp-service/examples/print-bmp.rs +++ b/crates/bmp-service/examples/print-bmp.rs @@ -25,9 +25,9 @@ use tracing::{debug, error}; use netcalyx_bmp_service::handle::BmpServerHandle; fn init_tracing() { - // Very simple setup at the moment to validate the instrumentation in the code - // is working in the future that should be configured automatically based on - // configuration options + // Very simple setup at the moment to validate the instrumentation in the + // code is working in the future that should be configured automatically + // based on configuration options let subscriber = tracing_subscriber::FmtSubscriber::builder() .with_max_level(tracing::Level::TRACE) .with_writer(std::io::stderr) diff --git a/crates/bmp-service/src/actor.rs b/crates/bmp-service/src/actor.rs index f8a8a1ea..1ed8b2f0 100644 --- a/crates/bmp-service/src/actor.rs +++ b/crates/bmp-service/src/actor.rs @@ -1,3 +1,4 @@ +// Copyright (C) 2026-present The NetCalyx Authors. // Copyright (C) 2026-present The NetGauze Authors. // // Licensed under the Apache License, Version 2.0 (the "License"); @@ -407,8 +408,8 @@ impl BmpActor { // Broadcast updated subscribers to all connections tasks if let Err(e) = self.subscribers_tx.send(self.subscribers.clone()) { - // This only happens if there are no active receivers (connection tasks), which - // is fine + // This only happens if there are no active receivers (connection + // tasks), which is fine debug!( actor_id = self.actor_id, local_addr = %self.local_addr, @@ -455,8 +456,8 @@ impl BmpActor { // Broadcast updated subscribers to all connections tasks if let Err(e) = self.subscribers_tx.send(self.subscribers.clone()) { - // This only happens if there are no active receivers (connection tasks), which - // is fine + // This only happens if there are no active receivers (connection + // tasks), which is fine debug!( actor_id = self.actor_id, local_addr = %self.local_addr, diff --git a/crates/bmp-service/src/actor/tests.rs b/crates/bmp-service/src/actor/tests.rs index 69c36502..07281717 100644 --- a/crates/bmp-service/src/actor/tests.rs +++ b/crates/bmp-service/src/actor/tests.rs @@ -1,3 +1,4 @@ +// Copyright (C) 2026-present The NetCalyx Authors. // Copyright (C) 2026-present The NetGauze Authors. // // Licensed under the Apache License, Version 2.0 (the "License"); @@ -276,7 +277,8 @@ async fn test_subscription_unsubscribe() { .await .expect("failed to send test BMP message"); - // The receiving channel should now be closed by the sender (actor dropped tx) + // The receiving channel should now be closed by the sender (actor dropped + // tx) timeout(Duration::from_millis(200), rx.recv()) .await .expect("Timed out waiting for channel closure") @@ -286,8 +288,8 @@ async fn test_subscription_unsubscribe() { #[tokio::test] #[tracing_test::traced_test] async fn test_subscription_updates_existing_connection() { - // This test ensures that an existing connection picks up new subscriptions via - // the broadcast channel + // This test ensures that an existing connection picks up new subscriptions + // via the broadcast channel let meter = opentelemetry::global::meter("test_subs_update"); let addr: SocketAddr = "127.0.0.1:0".parse().unwrap(); let (_join_handle, handle) = BmpActorHandle::new( diff --git a/crates/bmp-service/src/supervisor.rs b/crates/bmp-service/src/supervisor.rs index 26a27401..1db098da 100644 --- a/crates/bmp-service/src/supervisor.rs +++ b/crates/bmp-service/src/supervisor.rs @@ -1,3 +1,4 @@ +// Copyright (C) 2026-present The NetCalyx Authors. // Copyright (C) 2026-present The NetGauze Authors. // // Licensed under the Apache License, Version 2.0 (the "License"); @@ -154,8 +155,8 @@ impl BmpSupervisor { let awaited = futures::future::select_all(join_handles); join_handles = match awaited.await { (Ok(ret), _, rest) => { - // TODO(AH): Have some policy to allow to restart actors or terminate supervisor - // if failed + // TODO(AH): Have some policy to allow to restart actors or + // terminate supervisor if failed if let Err(err) = ret { error!(error = %err, "Actor terminated with error"); } diff --git a/crates/collector/benches/enrich_ipfix_packet.rs b/crates/collector/benches/enrich_ipfix_packet.rs index e436aefa..3484855f 100644 --- a/crates/collector/benches/enrich_ipfix_packet.rs +++ b/crates/collector/benches/enrich_ipfix_packet.rs @@ -1,3 +1,4 @@ +// Copyright (C) 2026-present The NetCalyx Authors. // Copyright (C) 2025-present The NetGauze Authors. // // Licensed under the Apache License, Version 2.0 (the "License"); @@ -172,8 +173,8 @@ fn create_enrichment_cache(peer_ip: IpAddr, cache_scale: u32) -> EnrichmentCache } } - // Add combined ingress+egress scopes for a small subset (only for obs_domain_id - // 10) + // Add combined ingress+egress scopes for a small subset (only for + // obs_domain_id 10) for ingress_if in 100..(110) { for egress_if in 200..(210) { cache.apply_enrichment(EnrichmentOperation::Upsert(UpsertPayload { diff --git a/crates/collector/src/flow/aggregation/aggregator/tests.rs b/crates/collector/src/flow/aggregation/aggregator/tests.rs index 562966ff..608c7112 100644 --- a/crates/collector/src/flow/aggregation/aggregator/tests.rs +++ b/crates/collector/src/flow/aggregation/aggregator/tests.rs @@ -1,3 +1,4 @@ +// Copyright (C) 2026-present The NetCalyx Authors. // Copyright (C) 2025-present The NetGauze Authors. // // Licensed under the Apache License, Version 2.0 (the "License"); @@ -1177,8 +1178,8 @@ fn test_explode_netflowv9_missing_fields() { #[test] fn test_aggregator_push_netflowv9_and_ipfix_different_flow_types() { - // Test that IPFIX and NetFlow v9 flows with same key fields are NOT aggregated - // together because they have different flow_type discriminants + // Test that IPFIX and NetFlow v9 flows with same key fields are NOT + // aggregated together because they have different flow_type discriminants let config = create_test_config( Box::new([ FieldRef::new(IE::sourceIPv4Address, 0), diff --git a/crates/collector/src/flow/enrichment/actor/tests.rs b/crates/collector/src/flow/enrichment/actor/tests.rs index 3a8a0e51..d7cb6a7c 100644 --- a/crates/collector/src/flow/enrichment/actor/tests.rs +++ b/crates/collector/src/flow/enrichment/actor/tests.rs @@ -1,3 +1,4 @@ +// Copyright (C) 2026-present The NetCalyx Authors. // Copyright (C) 2025-present The NetGauze Authors. // // Licensed under the Apache License, Version 2.0 (the "License"); @@ -314,7 +315,8 @@ fn test_enrich_netflowv9_with_cached_metadata() { }]), )); - // Create expected enriched flow — same structure with enrichment fields added + // Create expected enriched flow — same structure with enrichment fields + // added let expected_flow = FlowInfo::NetFlowV9(NetFlowV9Packet::new( 1000, unix_time, diff --git a/crates/collector/src/flow/enrichment/cache.rs b/crates/collector/src/flow/enrichment/cache.rs index 0dc5bdeb..1fc4f9a1 100644 --- a/crates/collector/src/flow/enrichment/cache.rs +++ b/crates/collector/src/flow/enrichment/cache.rs @@ -1,3 +1,4 @@ +// Copyright (C) 2026-present The NetCalyx Authors. // Copyright (C) 2025-present The NetGauze Authors. // // Licensed under the Apache License, Version 2.0 (the "License"); @@ -151,7 +152,8 @@ impl EnrichmentCache { Some(fields) => fields, }; - // Index incoming fields with FieldRef and store as WeightedField entries + // Index incoming fields with FieldRef and store as WeightedField + // entries let indexed_incoming: FxHashMap = FieldRef::map_fields(&incoming_fields, |field_ref, field| { (field_ref, WeightedField::new(weight, field.clone())) @@ -699,7 +701,8 @@ impl std::fmt::Display for PeerMetadata { format_scope_entries(f, 0, scope_fields, fields, &mut first_scope)?; } - // Format domain-specific scopes (sorted by obs_domain_id for consistency) + // Format domain-specific scopes (sorted by obs_domain_id for + // consistency) let mut sorted_domains: Vec<_> = self.domain_scoped.iter().collect(); sorted_domains.sort_by_key(|(obs_id, _)| *obs_id); diff --git a/crates/collector/src/flow/enrichment/cache/tests.rs b/crates/collector/src/flow/enrichment/cache/tests.rs index 1caf51ec..ebee9100 100644 --- a/crates/collector/src/flow/enrichment/cache/tests.rs +++ b/crates/collector/src/flow/enrichment/cache/tests.rs @@ -1,3 +1,4 @@ +// Copyright (C) 2026-present The NetCalyx Authors. // Copyright (C) 2025-present The NetGauze Authors. // // Licensed under the Apache License, Version 2.0 (the "License"); @@ -567,8 +568,8 @@ fn test_peer_metadata_get_enrichment_fields_multiple_scopes_some_matching() { WeightedField::new(100, Field::udpExID(29)), ); - // Domain-specific scoped fields that won't match (obs_domain_id = 20, no scope - // fields) + // Domain-specific scoped fields that won't match + // (obs_domain_id = 20, no scope fields) let mut specific_obs_id_nomatch_map = FxHashMap::default(); specific_obs_id_nomatch_map.insert( FieldRef::new(IE::internalAddressRealm, 0), @@ -644,8 +645,8 @@ fn test_peer_metadata_get_enrichment_fields_weight_priority() { ), ); - // Domain-specific scope with scope fields (obs_domain_id = 1000, selectorId = - // 1) + // Domain-specific scope with scope fields + // (obs_domain_id = 1000, selectorId = 1) let mut more_specific_fields = FxHashMap::default(); more_specific_fields.insert( FieldRef::new(IE::samplerName, 0), @@ -748,8 +749,8 @@ fn test_peer_metadata_get_enrichment_fields_same_weight_specificity_tiebreaker() WeightedField::new(100, Field::meteringProcessId(2000)), ); - // Domain-specific scope with scope fields (obs_domain_id = 1000, selectorId = - // 5) + // Domain-specific scope with scope fields + // (obs_domain_id = 1000, selectorId = 5) let mut more_specific_fields = FxHashMap::default(); more_specific_fields.insert( FieldRef::new(IE::applicationName, 0), diff --git a/crates/collector/src/flow/renormalization/logic.rs b/crates/collector/src/flow/renormalization/logic.rs index 05792f63..416fcc45 100644 --- a/crates/collector/src/flow/renormalization/logic.rs +++ b/crates/collector/src/flow/renormalization/logic.rs @@ -1,3 +1,4 @@ +// Copyright (C) 2026-present The NetCalyx Authors. // Copyright (C) 2026-present The NetGauze Authors. // // Licensed under the Apache License, Version 2.0 (the "License"); @@ -114,7 +115,8 @@ fn calculate_renormalization_factor( if let Some(alg) = params.selector_algorithm_304 { match alg { selectorAlgorithm::SystematiccountbasedSampling => { - // should have fields samplingPacketInterval and samplingPacketSpace + // should have fields samplingPacketInterval and + // samplingPacketSpace if let (Some(interval), Some(space)) = ( params.sampling_packet_interval_305, params.sampling_packet_space_306, @@ -323,7 +325,8 @@ fn renormalize_fields( stats.flows_processed.add(1, stats_tags); - // we expect records that have been already enriched with packet sampling IEs + // we expect records that have been already enriched with packet sampling + // IEs for field in &fields { match field { Field::samplingInterval(v) => params.sampling_interval_34 = Some(*v), @@ -384,9 +387,9 @@ pub(crate) fn renormalize( stats: &RenormalizationStats, stats_tags: &[KeyValue], ) -> FlowInfo { - // If there is any packet sampling information in the packet, then we adjust the - // flow packets and bytes and then add the isRenormalized boolean field to - // true. Otherwise, we leave the flow as is. + // If there is any packet sampling information in the packet, then + // we adjust the flow packets and bytes and add the isRenormalized + // boolean field to true. Otherwise, we leave the flow as it is. match info { FlowInfo::NetFlowV9(pkt) => { let sys_up_time = pkt.sys_up_time(); diff --git a/crates/collector/src/flow/renormalization/logic/tests.rs b/crates/collector/src/flow/renormalization/logic/tests.rs index be9e7e16..dab57b66 100644 --- a/crates/collector/src/flow/renormalization/logic/tests.rs +++ b/crates/collector/src/flow/renormalization/logic/tests.rs @@ -1,3 +1,4 @@ +// Copyright (C) 2026-present The NetCalyx Authors. // Copyright (C) 2026-present The NetGauze Authors. // // Licensed under the Apache License, Version 2.0 (the "License"); @@ -243,7 +244,8 @@ fn test_selector_algorithm_uniform_probabilistic_missing() { #[test] #[traced_test] fn test_selector_algorithm_unsupported() { - // 2 = Systematic time-based Sampling (not currently implemented in renormalize) + // 2 = Systematic time-based Sampling (not currently implemented in + // renormalize) let fields: Box<[Field]> = vec![Field::selectorAlgorithm( selectorAlgorithm::SystematictimebasedSampling, )] diff --git a/crates/collector/src/inputs/files/handlers.rs b/crates/collector/src/inputs/files/handlers.rs index ee6d9395..265fac8d 100644 --- a/crates/collector/src/inputs/files/handlers.rs +++ b/crates/collector/src/inputs/files/handlers.rs @@ -79,8 +79,8 @@ impl FilesLineHandler for FlowUpse )]) } LineChangeType::Removed => { - // If line was removed: generate delete payload from upsert payload (one-way - // conversion) + // If line was removed: generate delete payload from upsert + // payload (one-way conversion) let delete: crate::flow::enrichment::DeletePayload = upsert.into(); Ok(vec![crate::flow::enrichment::EnrichmentOperation::Delete( delete, @@ -142,8 +142,8 @@ impl FilesLineHandler for YangPushUpserts Ok(vec![crate::yang_push::EnrichmentOperation::Upsert(upsert)]) } LineChangeType::Removed => { - // If line was removed: generate delete payload from upsert payload (one-way - // conversion) + // If line was removed: generate delete payload from upsert + // payload (one-way conversion) let delete: crate::yang_push::DeletePayload = upsert.into(); Ok(vec![crate::yang_push::EnrichmentOperation::Delete(delete)]) } @@ -563,7 +563,8 @@ id=2:4200137808:1003 ip=192.168.100.1 out=127"#; .unwrap(); result.sort(); - // Should generate two operations: one for ingress VRF, one for egress VRF + // Should generate two operations: + // one for ingress VRF, one for egress VRF assert_eq!(result.len(), 2); let mut expected = vec![ diff --git a/crates/collector/src/inputs/flow_options/normalize.rs b/crates/collector/src/inputs/flow_options/normalize.rs index 1d615b07..cabe17f0 100644 --- a/crates/collector/src/inputs/flow_options/normalize.rs +++ b/crates/collector/src/inputs/flow_options/normalize.rs @@ -1,3 +1,4 @@ +// Copyright (C) 2026-present The NetCalyx Authors. // Copyright (C) 2025-present The NetGauze Authors. // // Licensed under the Apache License, Version 2.0 (the "License"); @@ -104,10 +105,11 @@ impl TryFrom for OptionsDataRecord { // peer IP address as the system identifier. } netflow::ScopeField::Interface(iface) => { - // In the case of interface scope we push both ingress and egress - // interfaces as scope fields. Thanks to the normalize_interface_type() - // they will be split into two IndexedDataRecord with the same interface - // ID but different ingress/egress specific fields. + // In the case of interface scope we push both ingress + // and egress interfaces as scope fields. Thanks to the + // normalize_interface_type() they will be split into + // two IndexedDataRecord with the same interface ID but + // different ingress/egress specific fields. scope_fields.extend([ Field::ingressInterface(iface.0), Field::egressInterface(iface.0), diff --git a/crates/collector/src/lib.rs b/crates/collector/src/lib.rs index bf4f224d..acfac916 100644 --- a/crates/collector/src/lib.rs +++ b/crates/collector/src/lib.rs @@ -481,8 +481,8 @@ pub async fn init_bmp_collection( } PublisherEndpoint::BmpKafkaAvro(config) => { for bmp_recv in &bmp_recvs { - // pass writer_id to the converter - // (workaround until we have an enrichment actor for bmp) + // pass writer_id to the converter (workaround + // until we have an enrichment actor for bmp) let mut config = config.clone(); config.avro_converter.writer_id = config.writer_id.clone(); diff --git a/crates/collector/src/main.rs b/crates/collector/src/main.rs index 0d319c8e..5792f697 100644 --- a/crates/collector/src/main.rs +++ b/crates/collector/src/main.rs @@ -282,13 +282,15 @@ fn main() -> anyhow::Result<()> { let runtime = runtime_builder.build()?; // Dedicated runtime for the OTEL metrics PeriodicReader so its export task - // doesn't compete with the collection/publishing pipeline for worker threads. + // doesn't compete with the collection/publishing pipeline for worker + // threads. let telemetry_runtime = tokio::runtime::Builder::new_multi_thread() .worker_threads(1) .enable_all() .build()?; let meter_provider = { - // `.enter()` makes the tokio::spawn done inside build() bind to telemetry_runtime. + // `.enter()` makes the tokio::spawn done inside build() bind to + // telemetry_runtime. let _guard = telemetry_runtime.enter(); init_open_telemetry(&config.telemetry).map_err(|err| anyhow!(err))? }; diff --git a/crates/collector/src/publishers/kafka_avro.rs b/crates/collector/src/publishers/kafka_avro.rs index 58251824..88ec7b89 100644 --- a/crates/collector/src/publishers/kafka_avro.rs +++ b/crates/collector/src/publishers/kafka_avro.rs @@ -1,3 +1,4 @@ +// Copyright (C) 2026-present The NetCalyx Authors. // Copyright (C) 2025-present The NetGauze Authors. // // Licensed under the Apache License, Version 2.0 (the "License"); @@ -372,7 +373,8 @@ where Err((err, rec)) => { match err { KafkaError::MessageProduction(RDKafkaErrorCode::QueueFull) => { - // Exponential backoff when the librdkafka is full + // Exponential backoff when the librdkafka is + // full if polling_interval > MAX_POLLING_INTERVAL { error!("Kafka polling interval exceeded, dropping record"); self.stats.error_send.add( diff --git a/crates/collector/src/publishers/kafka_yang.rs b/crates/collector/src/publishers/kafka_yang.rs index 30f44048..f6f7cd29 100644 --- a/crates/collector/src/publishers/kafka_yang.rs +++ b/crates/collector/src/publishers/kafka_yang.rs @@ -508,8 +508,8 @@ where None }; - // Load and register provided custom schemas - // (custom schemas are already extended with the telemetry-message schema) + // Load and register provided custom schemas (custom schemas + // are already extended with the telemetry-message schema) let mut schema_id_cache = HashMap::new(); for yang_lib_ref in custom_schemas.values() { diff --git a/crates/flow-pkt/build.rs b/crates/flow-pkt/build.rs index efc57587..db6a656f 100644 --- a/crates/flow-pkt/build.rs +++ b/crates/flow-pkt/build.rs @@ -1,3 +1,4 @@ +// Copyright (C) 2026-present The NetCalyx Authors. // Copyright (C) 2022-present The NetGauze Authors. // // Licensed under the Apache License, Version 2.0 (the "License"); @@ -201,7 +202,7 @@ fn get_vmware_config( 954, )]; - // // Protocol Numbers SubRegistry Path is either loaded from IANA or locally + // Protocol Numbers SubRegistry Path is either loaded from IANA or locally if poll_iana_registry { external_sub_registries.push(ExternalSubRegistrySource::new( RegistrySource::Http(PROTOCOL_NUMBERS_URL.to_string()), diff --git a/crates/flow-pkt/examples/ipfix-subregs.rs b/crates/flow-pkt/examples/ipfix-subregs.rs index d5873026..93a73e27 100644 --- a/crates/flow-pkt/examples/ipfix-subregs.rs +++ b/crates/flow-pkt/examples/ipfix-subregs.rs @@ -1,3 +1,4 @@ +// Copyright (C) 2026-present The NetCalyx Authors. // Copyright (C) 2023-present The NetGauze Authors. // // Licensed under the Apache License, Version 2.0 (the "License"); @@ -85,9 +86,9 @@ fn main() { 0x01, 0xd2, 0x00, 0x01, 0x01, 0xd3, 0x00, 0x01, 0x01, 0xf4, 0x00, 0x01 ] ); - // Deserialize the message from binary format (this will also add the Template - // to templates_map, otherwise the packet will be generated with all the - // default lengths) + // Deserialize the message from binary format + // (this will also add the Template to templates_map, otherwise + // the packet will be generated with all the default lengths) let (_, msg_back) = IpfixPacket::from_wire(Span::new(&buf), &mut templates_map).unwrap(); assert_eq!(ipfix_template, msg_back); diff --git a/crates/flow-pkt/examples/ipfix-vmware.rs b/crates/flow-pkt/examples/ipfix-vmware.rs index 20eda0f7..006d4a8b 100644 --- a/crates/flow-pkt/examples/ipfix-vmware.rs +++ b/crates/flow-pkt/examples/ipfix-vmware.rs @@ -1,3 +1,4 @@ +// Copyright (C) 2026-present The NetCalyx Authors. // Copyright (C) 2023-present The NetGauze Authors. // // Licensed under the Apache License, Version 2.0 (the "License"); @@ -81,9 +82,9 @@ fn main() { ] ); - // Deserialize the message from binary format (this will also add the Template - // to templates_map, otherwise the packet will be generated with all the - // default lengths) + // Deserialize the message from binary format + // (this will also add the Template to templates_map, otherwise + // the packet will be generated with all the default lengths) let (_, msg_back) = IpfixPacket::from_wire(Span::new(&buf), &mut templates_map).unwrap(); assert_eq!(ipfix_template, msg_back); diff --git a/crates/flow-pkt/src/codec.rs b/crates/flow-pkt/src/codec.rs index 93d38361..c1e9d742 100644 --- a/crates/flow-pkt/src/codec.rs +++ b/crates/flow-pkt/src/codec.rs @@ -1,3 +1,4 @@ +// Copyright (C) 2026-present The NetCalyx Authors. // Copyright (C) 2023-present The NetGauze Authors. // // Licensed under the Apache License, Version 2.0 (the "License"); @@ -172,8 +173,8 @@ fn parse_ipfix( FlowInfoCodecDecoderError::IpfixParsingError(error.error().clone()) } }; - // Make sure we advance the buffer far enough, so we don't get stuck on - // an error value. + // Make sure we advance the buffer far enough, + // so we don't get stuck on an error value. buf.advance(if length < 5 { 5 } else { length }); return Err(err); } @@ -204,10 +205,10 @@ fn parse_netflow_v9( FlowInfoCodecDecoderError::NetFlowV9ParingError(error.error().clone()) } }; - // Netflow v9 doesn't have a length component to tell us how many bytes - // should skip for the next packet. Sadly, our best bet is to clear the - // buffer and start over at the risk of discarding other good packets in - // the buffer. + // Netflow v9 doesn't have a length component to tell us how many + // bytes should skip for the next packet. Sadly, our best bet is to + // clear the buffer and start over at the risk of discarding other + // good packets in the buffer. buf.clear(); return Err(err); } @@ -221,16 +222,16 @@ impl Decoder for FlowInfoCodec { #[instrument(skip_all)] fn decode(&mut self, buf: &mut BytesMut) -> Result, Self::Error> { - // We're using IPFIX_HEADER_LENGTH as criteria to start parsing since it's - // smaller than NetFlow v9 header size. + // We're using IPFIX_HEADER_LENGTH as criteria to start parsing since + // it's smaller than NetFlow v9 header size. let header_length = IPFIX_HEADER_LENGTH as usize; if buf.len() < header_length { // We don't have enough data yet to start processing return Ok(None); } let version: u16 = NetworkEndian::read_u16(&buf[0..2]); - // Read the length (ipfix) or count (NetFlow v9), starting from after the - // version + // Read the length (ipfix) or count (NetFlow v9), starting from after + // the version let length = NetworkEndian::read_u16(&buf[2..4]) as usize; if buf.len() < length { // We still didn't read all the bytes for the message yet diff --git a/crates/flow-pkt/src/lib.rs b/crates/flow-pkt/src/lib.rs index eed9bf9c..3c639ed6 100644 --- a/crates/flow-pkt/src/lib.rs +++ b/crates/flow-pkt/src/lib.rs @@ -1,3 +1,4 @@ +// Copyright (C) 2026-present The NetCalyx Authors. // Copyright (C) 2022-present The NetGauze Authors. // // Licensed under the Apache License, Version 2.0 (the "License"); @@ -277,8 +278,8 @@ mod tests { assert!(!IE::ipv6ExtensionHeadersFull.supports_arithmetic_ops()); assert!(IE::mibObjectValueInteger.supports_arithmetic_ops()); assert!(IE::absoluteError.supports_arithmetic_ops()); - // numbers that are identifiers, flags, or have subregistries don't support - // arithmetic ops + // numbers that are identifiers, flags, or have subregistries don't + // support arithmetic ops assert!(!IE::ipClassOfService.supports_arithmetic_ops()); assert!(!IE::egressInterface.supports_arithmetic_ops()); assert!(!IE::forwardingStatus.supports_arithmetic_ops()); @@ -310,8 +311,8 @@ mod tests { assert!(IE::postMCastPacketDeltaCount.supports_bitwise_ops()); assert!(IE::ipv6ExtensionHeadersFull.supports_bitwise_ops()); assert!(IE::mibObjectValueInteger.supports_bitwise_ops()); - // numbers that are identifiers, flags, or have subregistries support bitwise - // ops + // numbers that are identifiers, flags, or have subregistries support + // bitwise ops assert!(IE::ipClassOfService.supports_bitwise_ops()); assert!(IE::egressInterface.supports_bitwise_ops()); assert!(IE::forwardingStatus.supports_bitwise_ops()); @@ -339,8 +340,8 @@ mod tests { assert!(IE::postMCastPacketDeltaCount.supports_comparison_ops()); assert!(!IE::ipv6ExtensionHeadersFull.supports_comparison_ops()); assert!(IE::mibObjectValueInteger.supports_comparison_ops()); - // numbers that are identifiers, flags, or have subregistries support comparison - // ops + // numbers that are identifiers, flags, or have subregistries support + // comparison ops assert!(IE::ipClassOfService.supports_comparison_ops()); assert!(IE::egressInterface.supports_comparison_ops()); assert!(IE::forwardingStatus.supports_comparison_ops()); diff --git a/crates/flow-pkt/src/wire/deserializer/ipfix.rs b/crates/flow-pkt/src/wire/deserializer/ipfix.rs index 333189a2..cfe3ec10 100644 --- a/crates/flow-pkt/src/wire/deserializer/ipfix.rs +++ b/crates/flow-pkt/src/wire/deserializer/ipfix.rs @@ -1,3 +1,4 @@ +// Copyright (C) 2026-present The NetCalyx Authors. // Copyright (C) 2023-present The NetGauze Authors. // // Licensed under the Apache License, Version 2.0 (the "License"); @@ -213,10 +214,10 @@ impl<'a> ReadablePduWithOneInput<'a, &mut TemplatesMap, LocatedSetParsingError<' } IPFIX_OPTIONS_TEMPLATE_SET_ID => { let mut option_templates = vec![]; - // THE RFC is not super clear about - // length allowed in the Options - // Template set. Like Wireshark implementation, we assume anything - // less than 4-octets (min field size) is padding + // THE RFC is not super clear about length allowed in the + // Options Template set. Like Wireshark implementation, + // we assume anything less than 4-octets (min field size) + // is padding while buf.len() > 3 { // let (t, option_template) = // parse_into_located_one_input(buf, templates_map)?; @@ -248,7 +249,8 @@ impl<'a> ReadablePduWithOneInput<'a, &mut TemplatesMap, LocatedSetParsingError<' SetParsingError::NoTemplateDefinedFor(id), ))); }; - // since we could have vlen fields, we can only state a min_record_len here + // since we could have vlen fields, we can only state a + // min_record_len here let min_record_length = template .scope_fields_specs .iter() @@ -288,7 +290,8 @@ impl<'a> ReadablePduWithOneInput<'a, &mut TemplatesMap, LocatedSetParsingError<' buf = t; } - // We can safely unwrap DataSetId here since we already checked the range + // We can safely unwrap DataSetId here since we already checked + // the range Set::Data { id: DataSetId::new(id).unwrap(), records: records.into_boxed_slice(), @@ -356,8 +359,8 @@ impl<'a> ) -> IResult, Self, LocatedOptionsTemplateRecordParsingError<'a>> { let input = buf; let (buf, template_id) = be_u16(buf)?; - // from RFC7011: Each Template Record is given a unique Template ID in the range - // 256 to 65535. + // from RFC7011: Each Template Record is given a unique Template ID in + // the range 256 to 65535. if template_id < 256 { return Err(nom::Err::Error( LocatedOptionsTemplateRecordParsingError::new( @@ -500,8 +503,8 @@ impl<'a> ReadablePduWithOneInput<'a, &mut TemplatesMap, LocatedTemplateRecordPar ) -> IResult, Self, LocatedTemplateRecordParsingError<'a>> { let input = buf; let (buf, template_id) = be_u16(buf)?; - // from RFC7011: Each Template Record is given a unique Template ID in the range - // 256 to 65535. + // from RFC7011: Each Template Record is given a unique Template ID in + // the range 256 to 65535. if template_id < 256 { return Err(nom::Err::Error(LocatedTemplateRecordParsingError::new( input, diff --git a/crates/flow-pkt/src/wire/deserializer/netflow.rs b/crates/flow-pkt/src/wire/deserializer/netflow.rs index 119c80ce..34362fab 100644 --- a/crates/flow-pkt/src/wire/deserializer/netflow.rs +++ b/crates/flow-pkt/src/wire/deserializer/netflow.rs @@ -1,3 +1,4 @@ +// Copyright (C) 2026-present The NetCalyx Authors. // Copyright (C) 2023-present The NetGauze Authors. // // Licensed under the Apache License, Version 2.0 (the "License"); @@ -226,9 +227,10 @@ impl<'a> ReadablePduWithOneInput<'a, &mut TemplatesMap, LocatedSetParsingError<' } NETFLOW_OPTIONS_TEMPLATE_SET_ID => { let mut option_templates = vec![]; - // THE RFC is not super clear about padding length allowed in the Options - // Template set. Like Wireshark implementation, we assume anything - // less than 4-octets (min field size) is padding + // THE RFC is not super clear about padding length allowed in + // the Options Template set. Like Wireshark implementation, + // we assume anything less than 4-octets (min field size) + // is padding while buf.len() > 3 { let (t, option_template) = match OptionsTemplateRecord::from_wire(buf, templates_map) { @@ -285,7 +287,8 @@ impl<'a> ReadablePduWithOneInput<'a, &mut TemplatesMap, LocatedSetParsingError<' }; // buf could be a non zero value for padding check_padding_value(buf)?; - // We can safely unwrap DataSetId here since we already checked the range + // We can safely unwrap DataSetId here since we already checked + // the range Set::Data { id: DataSetId::new(id).unwrap(), records: records.into_boxed_slice(), @@ -358,8 +361,8 @@ impl<'a> ) -> IResult, Self, LocatedOptionsTemplateRecordParsingError<'a>> { let input = buf; let (buf, template_id) = be_u16(buf)?; - // from RFC7011: Each Template Record is given a unique Template ID in the range - // 256 to 65535. + // from RFC7011: Each Template Record is given a unique Template ID in + // the range 256 to 65535. if template_id < 256 { return Err(nom::Err::Error( LocatedOptionsTemplateRecordParsingError::new( @@ -441,8 +444,8 @@ impl<'a> ReadablePduWithOneInput<'a, &mut TemplatesMap, LocatedTemplateRecordPar ) -> IResult, Self, LocatedTemplateRecordParsingError<'a>> { let input = buf; let (buf, template_id) = be_u16(buf)?; - // from RFC7011: Each Template Record is given a unique Template ID in the range - // 256 to 65535. + // from RFC7011: Each Template Record is given a unique Template ID in + // the range 256 to 65535. if template_id < 256 { return Err(nom::Err::Error(LocatedTemplateRecordParsingError::new( input, diff --git a/crates/flow-pkt/src/wire/tests/ipfix.rs b/crates/flow-pkt/src/wire/tests/ipfix.rs index f4b325c1..f73bce13 100644 --- a/crates/flow-pkt/src/wire/tests/ipfix.rs +++ b/crates/flow-pkt/src/wire/tests/ipfix.rs @@ -1,3 +1,4 @@ +// Copyright (C) 2026-present The NetCalyx Authors. // Copyright (C) 2023-present The NetGauze Authors. // // Licensed under the Apache License, Version 2.0 (the "License"); @@ -1794,10 +1795,10 @@ fn test_octet_array_variable_length() -> Result<(), IpfixPacketWritingError> { #[test] fn test_padding_min_length_issue_360() -> Result<(), IpfixPacketWritingError> { - // data packet with one padding byte after an u8 field, and 3 variable lengths - // strings. This test ensures that the padding byte is correctly handled - // when calculating the min record value and the single padding octet is not - // considered a new data record. + // data packet with one padding byte after an u8 field, and 3 variable + // lengths strings. This test ensures that the padding byte is correctly + // handled when calculating the min record value and the single padding + // octet is not considered a new data record. let data_wire = [ 0x00, 0x0a, 0x00, 0x51, 0x69, 0x49, 0x2a, 0x58, 0x00, 0x00, 0x00, 0x0a, 0x00, 0x00, 0x00, 0x00, 0x01, 0xf4, 0x00, 0x41, 0xee, 0x07, 0x73, 0x6f, 0x6d, 0x65, 0x2d, 0x69, 0x64, 0x17, diff --git a/crates/flow-pkt/src/wire/tests/netflow.rs b/crates/flow-pkt/src/wire/tests/netflow.rs index 3733dd45..81757bfa 100644 --- a/crates/flow-pkt/src/wire/tests/netflow.rs +++ b/crates/flow-pkt/src/wire/tests/netflow.rs @@ -1,3 +1,4 @@ +// Copyright (C) 2026-present The NetCalyx Authors. // Copyright (C) 2023-present The NetGauze Authors. // // Licensed under the Apache License, Version 2.0 (the "License"); @@ -751,8 +752,8 @@ fn test_zero_length_fields() { 0, 0, 91, 0, 0, 0, 0, 91, 0, 0, 4, 0, 0, 0, 0, 32, 0, 0, ]; let mut templates_map = HashMap::new(); - // The test here will produce invalid packet, but what we are testing for is not - // crashing due to divide by zero + // The test here will produce invalid packet, but what we are testing for is + // not crashing due to divide by zero let ret = NetFlowV9Packet::from_wire(Span::new(&good_template_wire), &mut templates_map); assert!(ret.is_err()); } @@ -780,8 +781,8 @@ fn test_records_len_larger_than_count() { 123, 123, 123, 123, 255, 0, 0, ]; let mut templates_map = HashMap::new(); - // The test here will produce invalid packet, but what we are testing for is not - // crashing due subtracting count of records from the templates + // The test here will produce invalid packet, but what we are testing for is + // not crashing due subtracting count of records from the templates let ret = NetFlowV9Packet::from_wire(Span::new(&good_template_wire), &mut templates_map); assert!(ret.is_err()); } diff --git a/crates/flow-pkt/src/wire/tests/pcap_tests.rs b/crates/flow-pkt/src/wire/tests/pcap_tests.rs index d0e16918..2d85228a 100644 --- a/crates/flow-pkt/src/wire/tests/pcap_tests.rs +++ b/crates/flow-pkt/src/wire/tests/pcap_tests.rs @@ -1,3 +1,4 @@ +// Copyright (C) 2026-present The NetCalyx Authors. // Copyright (C) 2024-present The NetGauze Authors. // // Licensed under the Apache License, Version 2.0 (the "License"); @@ -93,8 +94,8 @@ fn test_flow_pcap(overwrite: bool, pcap_path: PathBuf) { serde_json::to_string(&msg).expect("Couldn't serialize Flow message to json") } Ok(None) => { - // packet is fragmented, need to read the next PDU first before attempting to - // deserialize it + // packet is fragmented, need to read the next PDU first + // before attempting to deserialize it break; } Err(err) => serde_json::to_string(&err) diff --git a/crates/flow-service/examples/actor-example.rs b/crates/flow-service/examples/actor-example.rs index bb780669..55a6d857 100644 --- a/crates/flow-service/examples/actor-example.rs +++ b/crates/flow-service/examples/actor-example.rs @@ -19,9 +19,9 @@ use std::time::Duration; use tracing::{debug, error, info}; fn init_tracing() { - // Very simple setup at the moment to validate the instrumentation in the code - // is working in the future that should be configured automatically based on - // configuration options + // Very simple setup at the moment to validate the instrumentation in the + // code is working in the future that should be configured automatically + // based on configuration options let subscriber = tracing_subscriber::FmtSubscriber::builder() .with_max_level(tracing::Level::DEBUG) .with_writer(std::io::stderr) diff --git a/crates/flow-service/examples/actors-example.rs b/crates/flow-service/examples/actors-example.rs index 7ef5fca9..f23d2d93 100644 --- a/crates/flow-service/examples/actors-example.rs +++ b/crates/flow-service/examples/actors-example.rs @@ -21,9 +21,9 @@ use std::time::Duration; use tracing::{debug, error, info}; fn init_tracing() { - // Very simple setup at the moment to validate the instrumentation in the code - // is working in the future that should be configured automatically based on - // configuration options + // Very simple setup at the moment to validate the instrumentation in the + // code is working in the future that should be configured automatically + // based on configuration options let subscriber = tracing_subscriber::FmtSubscriber::builder() .with_max_level(tracing::Level::DEBUG) .with_writer(std::io::stderr) diff --git a/crates/flow-service/examples/print-flow.rs b/crates/flow-service/examples/print-flow.rs index 25818850..c7ada083 100644 --- a/crates/flow-service/examples/print-flow.rs +++ b/crates/flow-service/examples/print-flow.rs @@ -24,9 +24,9 @@ use tokio_util::codec::{BytesCodec, Decoder}; use tokio_util::udp::UdpFramed; fn init_tracing() { - // Very simple setup at the moment to validate the instrumentation in the code - // is working in the future that should be configured automatically based on - // configuration options + // Very simple setup at the moment to validate the instrumentation in the + // code is working in the future that should be configured automatically + // based on configuration options let subscriber = tracing_subscriber::FmtSubscriber::builder() .with_max_level(tracing::Level::TRACE) .with_writer(std::io::stderr) @@ -47,9 +47,10 @@ async fn main() -> Result<(), Box while let Some(next) = stream.next().await { match next { Ok((mut buf, addr)) => { - // If we haven't seen the client before, create a new FlowInfoCodec for it. - // FlowInfoCodec handles the decoding/encoding of packets and caches - // the templates learned from the client + // If we haven't seen the client before, create a new + // FlowInfoCodec for it. FlowInfoCodec handles the + // decoding/encoding of packets and caches the templates + // learned from the client let result = clients .entry(addr) .or_insert(FlowInfoCodec::default()) diff --git a/crates/flow-service/src/flow_actor.rs b/crates/flow-service/src/flow_actor.rs index fcc2c186..ebd37110 100644 --- a/crates/flow-service/src/flow_actor.rs +++ b/crates/flow-service/src/flow_actor.rs @@ -1,3 +1,4 @@ +// Copyright (C) 2026-present The NetCalyx Authors. // Copyright (C) 2024-present The NetGauze Authors. // // Licensed under the Apache License, Version 2.0 (the "License"); @@ -341,9 +342,9 @@ impl FlowCollectorActor { /// decoding error), it logs the error and returns None. pub fn decode_pkt(&mut self, next: (BytesMut, SocketAddr)) -> Option { let (mut buf, addr) = next; - // If we haven't seen the client before, create a new FlowInfoCodec for it. - // FlowInfoCodec handles the decoding/encoding of packets and caches - // the templates learned from the client + // If we haven't seen the client before, create a new FlowInfoCodec for + // it. FlowInfoCodec handles the decoding/encoding of packets and caches + // the templates learned from the client. let mut attrs = vec![ opentelemetry::KeyValue::new("netcalyx.flow.actor", format!("{}", self.actor_id)), opentelemetry::KeyValue::new("network.peer.address", format!("{}", addr.ip())), @@ -421,8 +422,8 @@ impl FlowCollectorActor { sent_counter: opentelemetry::metrics::Counter, drop_counter: opentelemetry::metrics::Counter, ) { - // The send operation is bounded by timeout period to avoid blocking on a slow - // subscriber. + // The send operation is bounded by timeout period to avoid blocking on + // a slow subscriber. let ref_clone = pkt.clone(); let drop_counter_clone = drop_counter.clone(); let timeout_ret = tokio::time::timeout(timeout, async move { @@ -540,7 +541,8 @@ impl FlowCollectorActor { ); send_handlers.push(send_handler); } - // Avoid blocking on sending the packet to the subscribers, and focus on + // Avoid blocking on sending the packet to the subscribers, + // and focus on futures::future::join_all(send_handlers).await; } } @@ -841,7 +843,8 @@ impl FlowCollectorActor { )); } }; - // Get the local address of the socket, handy in cases where the port is 0 + // Get the local address of the socket, handy in cases where the port is + // 0 self.socket_addr = socket.local_addr().map_err(|err| { FlowCollectorActorError::GetLocalAddressError(self.actor_id, socket_addr, err) })?; @@ -1080,8 +1083,8 @@ impl FlowCollectorActorHandle { duration: Duration, ) -> Result, FlowCollectorActorHandleError> { let (tx, mut rx) = mpsc::channel(self.cmd_buffer_size); - // If the command fails, the recv after will fail, no need to double handle the - // error + // If the command fails, the recv after will fail, + // no need to double handle the error self.cmd_tx .send(FlowCollectorActorCommand::PurgeUnusedPeers(duration, tx)) .await diff --git a/crates/flow-service/src/flow_supervisor.rs b/crates/flow-service/src/flow_supervisor.rs index 9fa45ddf..eb82ba72 100644 --- a/crates/flow-service/src/flow_supervisor.rs +++ b/crates/flow-service/src/flow_supervisor.rs @@ -1,3 +1,4 @@ +// Copyright (C) 2026-present The NetCalyx Authors. // Copyright (C) 2024-present The NetGauze Authors. // // Licensed under the Apache License, Version 2.0 (the "License"); @@ -137,8 +138,8 @@ impl FlowCollectorsSupervisorActor { let awaited = futures::future::select_all(join_handles); join_handles = match awaited.await { (Ok(ret), _, rest) => { - // TODO(AH): Have some policy to allow to restart actors or terminate supervisor - // if failed + // TODO(AH): Have some policy to allow to restart actors or + // terminate supervisor if failed if let Err(err) = ret { error!("[Supervisor] actor terminated with error: {err}"); } @@ -695,8 +696,8 @@ mod test { assert_eq!(unsubscribe_results.len(), 3); assert!(unsubscribe_results.iter().all(|r| r.is_some())); tokio::time::sleep(Duration::from_secs(1)).await; - // Try to receive a message (should return None denoting channel is closed as - // we've unsubscribed) + // Try to receive a message (should return None denoting channel is + // closed as we've unsubscribed) let timeout_result = timeout(Duration::from_secs(1), pkt_rx.recv()).await; assert!(matches!(timeout_result, Ok(Err(_)))); diff --git a/crates/flow-service/src/lib.rs b/crates/flow-service/src/lib.rs index 99df5706..6984748a 100644 --- a/crates/flow-service/src/lib.rs +++ b/crates/flow-service/src/lib.rs @@ -1,3 +1,4 @@ +// Copyright (C) 2026-present The NetCalyx Authors. // Copyright (C) 2023-present The NetGauze Authors. // // Licensed under the Apache License, Version 2.0 (the "License"); @@ -77,8 +78,8 @@ pub fn new_udp_reuse_port( udp_sock.set_nonblocking(true)?; // Binding a socket to a device or VRF is platform specific operation, // hence we guard it for only selected subset of target platforms. - // The first cfg block filter for all supported platforms to avoid Clippy errors - // on unused `name` for the unsupported platforms. + // The first cfg block filters for all supported platforms to avoid + // Clippy errors on unused `name` for the unsupported platforms. #[cfg(any( target_os = "ios", target_os = "macos", diff --git a/crates/ipfix-code-generator/src/xml_parsers/sub_registries.rs b/crates/ipfix-code-generator/src/xml_parsers/sub_registries.rs index 68e93f00..4ce6fe84 100644 --- a/crates/ipfix-code-generator/src/xml_parsers/sub_registries.rs +++ b/crates/ipfix-code-generator/src/xml_parsers/sub_registries.rs @@ -1,3 +1,4 @@ +// Copyright (C) 2026-present The NetCalyx Authors. // Copyright (C) 2022-present The NetGauze Authors. // // Licensed under the Apache License, Version 2.0 (the "License"); @@ -188,8 +189,8 @@ pub fn parse_val_name_desc_u8_registry(node: &Node<'_, '_>) -> (u16, Vec Result<(), Box> { use tracing_subscriber::{EnvFilter, fmt}; // Set up the log -> tracing bridge first - // tracing_log::LogTracer::init().expect("Failed to initialize tracing logger"); + // tracing_log::LogTracer::init().expect("Failed to initialize tracing + // logger"); let env_filter = EnvFilter::try_from_default_env() .or_else(|_| EnvFilter::try_new("info")) diff --git a/crates/netconf-proto/src/client.rs b/crates/netconf-proto/src/client.rs index deb5893e..b0541f29 100644 --- a/crates/netconf-proto/src/client.rs +++ b/crates/netconf-proto/src/client.rs @@ -550,8 +550,9 @@ impl NetConfSshClient { return self.fetch_module_rpc(name, None).await; }; - // `module_cache` is a cheap-to-clone handle; clone it so the fetch below - // can borrow `&mut self` without conflicting with the cache borrow. + // `module_cache` is a cheap-to-clone handle; clone it so the fetch + // below can borrow `&mut self` without conflicting with the + // cache borrow. let cache = self.module_cache.clone(); loop { match cache.begin_fetch(name, version) { @@ -570,15 +571,17 @@ impl NetConfSshClient { if let Some(text) = waiter.wait().await { return Ok(text); } - // The leader failed; retry — we become a new leader or waiter. + // The leader failed; retry — we become a new leader or + // waiter. debug!( "[{}] leader fetch for `{name}` revision {version} failed, retrying", self.peer ); } ModuleFetch::Lead(lease) => { - // We are the leader: perform the RPC. On success publish the - // text to waiters; on error the lease drops, freeing them. + // We are the leader: perform the RPC. + // On success publish the text to waiters; + // on error the lease drops, freeing them. let text = self.fetch_module_rpc(name, Some(version)).await?; lease.fulfil(Arc::clone(&text)); return Ok(text); @@ -892,8 +895,8 @@ impl NetConfSshClient { "[{}] Raw response for subscription {id}: `{data}`", self.peer ); - // Parse the response streams if any returned, filters if any returned - // and then the subscription details + // Parse the response streams if any returned, filters if any + // returned and then the subscription details let mut reader = NsReader::from_str(data); reader.config_mut().trim_text(true); let mut parser = crate::xml_utils::XmlParser::new(reader)?; diff --git a/crates/netconf-proto/src/codec.rs b/crates/netconf-proto/src/codec.rs index 38d97314..fb2c46b1 100644 --- a/crates/netconf-proto/src/codec.rs +++ b/crates/netconf-proto/src/codec.rs @@ -286,7 +286,8 @@ impl Decoder for SshCodec { // Parse chunk size let chunk_size_slice = &src[size_start..size_end]; let chunk_size_str = std::str::from_utf8(chunk_size_slice)?; - // RFC 6242 chunk-size: at least one digit, leading zeros are prohibited + // RFC 6242 chunk-size: at least one digit, + // leading zeros are prohibited if chunk_size_str.is_empty() || chunk_size_str.starts_with('0') { return Err(SshCodecError::IO(std::io::Error::new( std::io::ErrorKind::InvalidData, diff --git a/crates/netconf-proto/src/lib.rs b/crates/netconf-proto/src/lib.rs index 9b4e34fe..bc969738 100644 --- a/crates/netconf-proto/src/lib.rs +++ b/crates/netconf-proto/src/lib.rs @@ -131,8 +131,8 @@ mod tests { "Expecting:\n{expected:#?}\nparsed:\n{parsed:#?}" ); - // Check after we serialize the test value we can deserialize back the same - // value + // Check after we serialize the test value we can deserialize back the + // same value let writer = quick_xml::writer::Writer::new(io::Cursor::new(Vec::new())); let mut writer = XmlWriter::new(writer); parsed.xml_serialize(&mut writer)?; diff --git a/crates/netconf-proto/src/protocol.rs b/crates/netconf-proto/src/protocol.rs index e812b67e..1b7055ec 100644 --- a/crates/netconf-proto/src/protocol.rs +++ b/crates/netconf-proto/src/protocol.rs @@ -3248,7 +3248,8 @@ mod tests { #[test] fn test_yang_library_rfc8525() -> Result<(), ParsingError> { - // RFC 8525 Appendix C - Example YANG Library Instance for an Advanced Server + // RFC 8525 Appendix C - Example YANG Library Instance for an Advanced + // Server let library_str = r#" state-only-modules diff --git a/crates/netconf-proto/src/xml_utils.rs b/crates/netconf-proto/src/xml_utils.rs index 7d671923..d2e1a465 100644 --- a/crates/netconf-proto/src/xml_utils.rs +++ b/crates/netconf-proto/src/xml_utils.rs @@ -664,9 +664,10 @@ impl<'a, R: io::BufRead> XmlParser<'a, R> { return Err(ParsingError::Eof); } - // Record prefix usage on Start/Empty. We resolve NOW so that bindings - // declared on inner elements are captured even after those elements - // close and their bindings are popped from the resolver stack. + // Record prefix usage on Start/Empty. We resolve NOW so that + // bindings declared on inner elements are captured even after + // those elements close and their bindings are popped from the + // resolver stack. if let Event::Start(e) | Event::Empty(e) = self.peek() { Self::record_element_usage(e, &self.ns_reader, &mut namespaces); } @@ -791,8 +792,8 @@ impl<'a, R: io::BufRead> XmlParser<'a, R> { out.insert(prefix, String::from_utf8_lossy(uri).into_owned()); } - // Attributes: only prefixed ones carry a namespace; xmlns* are bindings, - // not usages, so skip them. + // Attributes: only prefixed ones carry a namespace; + // xmlns* are bindings, not usages, so skip them. for attr in e.attributes().flatten() { let key = attr.key.as_ref(); if key == b"xmlns" || key.starts_with(b"xmlns:") { @@ -1171,7 +1172,8 @@ mod tests { let result = parser.maybe_open(Some(Namespace(b"https://example.com")), "wrong"); assert_eq!(result, expected); - // check pointer didn't move after the `maybe_open` didn't return anything + // check pointer didn't move after the `maybe_open` didn't return + // anything assert_eq!( parser.peek(), &Event::Start( diff --git a/crates/netconf-proto/src/yang_push/filters.rs b/crates/netconf-proto/src/yang_push/filters.rs index f2b226dd..c0b7c992 100644 --- a/crates/netconf-proto/src/yang_push/filters.rs +++ b/crates/netconf-proto/src/yang_push/filters.rs @@ -117,8 +117,8 @@ impl XmlDeserialize<'_, Filters> for Filters { let selection_filter = SelectionFilter::xml_deserialize(parser)?; selection_filters.push(selection_filter); } else { - // Could be an IETF or vendor-specific extension that we don't understand, - // skip it + // Could be an IETF or vendor-specific extension that we don't + // understand, skip it parser.skip()?; } parser.skip_text()?; diff --git a/crates/netconf-proto/src/yang_push/tests.rs b/crates/netconf-proto/src/yang_push/tests.rs index 7eed95d7..fd28b193 100644 --- a/crates/netconf-proto/src/yang_push/tests.rs +++ b/crates/netconf-proto/src/yang_push/tests.rs @@ -323,7 +323,8 @@ fn roundtrip_within_wrapper< "Deserialized value differs:\n expected: {expected:#?}\n parsed: {parsed:#?}" ); - // 2. Serialize the expected value back out, then re-parse to confirm round-trip + // 2. Serialize the expected value back out, then re-parse to confirm + // round-trip let serialized = serialize_within_wrapper(&expected, wrapper_ns, wrapper_tag); let reparsed: T = parse_within_wrapper(&serialized, wrapper_ns, wrapper_tag) .expect("re-deserialization from serialized output failed"); diff --git a/crates/netconf-proto/src/yanglib.rs b/crates/netconf-proto/src/yanglib.rs index 9b71c717..91aa3837 100644 --- a/crates/netconf-proto/src/yanglib.rs +++ b/crates/netconf-proto/src/yanglib.rs @@ -252,8 +252,8 @@ impl YangLibrary { topo_sort ); } - // safe to unwrap since we constructed the graph above and checked the root - // module exists + // safe to unwrap since we constructed the graph above and checked the + // root module exists let root_index = root_index.unwrap(); let mut supplied_references: HashMap< &str, @@ -406,9 +406,10 @@ impl YangLibrary { let registered_schema_result = client.register_schema(&subject, schema, false).await; let registered_schema = match registered_schema_result { Ok(registered_schema) => { - // version number is only returned from schema registry 7.4 and higher - // older versions don't return the version number, thus we need to make - // more calls to the schema registry to obtain the version number. + // version number is only returned from schema registry 7.4 + // and higher, older versions don't return the version number, + // thus we need to make more calls to the schema registry to + // obtain the version number. if registered_schema.version.is_none() { let schema = client .get_by_subject_and_id(Some(&subject), registered_schema.id.unwrap(), None) @@ -964,9 +965,10 @@ impl ModuleSet { modules: Vec, import_only_modules: Vec, ) -> Self { - // TODO: we silently keep the last entry if `modules` has duplicate names, - // we should consider returning an error or warning if duplicates are found - // (RFC 8525 `module` list is keyed by name only, so duplicates shouldn't occur). + // TODO: we silently keep the last entry if `modules` has duplicate + // names, we should consider returning an error or warning if + // duplicates are found (RFC 8525 `module` list is keyed by name + // only, so duplicates shouldn't occur). let modules_map = IndexMap::from_iter(modules.into_iter().map(|m| (m.name.clone(), m))); let mut import_only_map = IndexMap::with_capacity(import_only_modules.len()); for import_only in import_only_modules { @@ -1949,8 +1951,8 @@ impl ModuleSetBuilder { where C: BackwardCompatibilityChecker, { - // Check the module which is the submodule is attached to is already defined in - // the module set. + // Check the module which is the submodule is attached to is already + // defined in the module set. let module = if let Some(module) = self.module_set.modules.get(module_name) { module } else { @@ -2798,7 +2800,8 @@ mod tests { #[test] fn test_rfc8525_appendix_c_advanced_server_serde() { - // RFC 8525 Appendix C - Example YANG Library Instance for an Advanced Server + // RFC 8525 Appendix C - Example YANG Library Instance for an Advanced + // Server let xml = r#" diff --git a/crates/netconf-proto/src/yangparser.rs b/crates/netconf-proto/src/yangparser.rs index d5f8fe12..64a2512d 100644 --- a/crates/netconf-proto/src/yangparser.rs +++ b/crates/netconf-proto/src/yangparser.rs @@ -233,8 +233,9 @@ impl<'a> YangParser<'a> { || self.match_keyword("description") || self.match_keyword("reference") { - // once we reach the meta-stmt we can stop parsing for imports/includes - // since they are not allowed in meta-stmts or any statement after them. + // once we reach the meta-stmt we can stop parsing for + // imports/includes since they are not allowed in meta-stmts + // or any statement after them. break; } else { // Skip one token at a time instead of entire statements diff --git a/crates/pcap-decoder/src/handlers/bgp.rs b/crates/pcap-decoder/src/handlers/bgp.rs index 4cb9d4c5..7b6eb1f0 100644 --- a/crates/pcap-decoder/src/handlers/bgp.rs +++ b/crates/pcap-decoder/src/handlers/bgp.rs @@ -1,3 +1,4 @@ +// Copyright (C) 2026-present The NetCalyx Authors. // Copyright (C) 2025-present The NetGauze Authors. // // Licensed under the Apache License, Version 2.0 (the "License"); @@ -82,9 +83,10 @@ impl ProtocolHandler<(BgpMessage, BgpParsingIgnoredErrors), BgpCodec, BgpCodecDe DecodeOutcome::Success(m) => { let (flow_key, (bgp_message, bgp_parsing_error)) = m; if !bgp_parsing_error.eq(&BgpParsingIgnoredErrors::default()) { - // the bgp message was parsed with some ignored errors, we will not serialize it - // we will report that some ignored errors were found and that this behavior - // by the CLI tool is not expected + // the bgp message was parsed with some ignored errors, + // we will not serialize it we will report that some ignored + // errors were found and that this behavior by the CLI tool + // is not expected return Ok(serde_json::Value::String("Encountered BGP parsing errors that were ignored during the decoding of the bgp message, this behaviour is not expected, please file a bug report to the developers".to_string())); } serialize_success(flow_key, bgp_message) @@ -175,7 +177,8 @@ mod tests { &mut exporter_peers, ); assert!(result1.is_none()); - // The buffer for this flow key should now contain the first part, so not empty + // The buffer for this flow key should now contain the first part, + // so not empty assert!(!exporter_peers.get(&flow_key).unwrap().1.is_empty()); // Second packet completes it @@ -273,8 +276,8 @@ mod tests { IpAddr::V4(Ipv4Addr::new(10, 0, 0, 2)), 179, ); - // The packet data contains a BGP message with some errors that can potentially - // be ignored + // The packet data contains a BGP message with some errors that can + // potentially be ignored let packet_data = [ 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0x00, 0x59, 0x02, 0x00, 0x00, 0x00, 0x30, 0x40, 0x01, 0x01, diff --git a/crates/pcap-decoder/src/handlers/bmp.rs b/crates/pcap-decoder/src/handlers/bmp.rs index ee19f978..f5cc88f0 100644 --- a/crates/pcap-decoder/src/handlers/bmp.rs +++ b/crates/pcap-decoder/src/handlers/bmp.rs @@ -1,3 +1,4 @@ +// Copyright (C) 2026-present The NetCalyx Authors. // Copyright (C) 2025-present The NetGauze Authors. // // Licensed under the Apache License, Version 2.0 (the "License"); @@ -157,7 +158,8 @@ mod tests { &mut exporter_peers, ); assert!(result1.is_none()); - // The buffer for this flow key should now contain the first part, so not empty + // The buffer for this flow key should now contain the first part, + // so not empty assert!(!exporter_peers.get(&flow_key).unwrap().1.is_empty()); let result2 = handler.decode( diff --git a/crates/pcap-decoder/src/handlers/flow.rs b/crates/pcap-decoder/src/handlers/flow.rs index e9209ca3..05622e54 100644 --- a/crates/pcap-decoder/src/handlers/flow.rs +++ b/crates/pcap-decoder/src/handlers/flow.rs @@ -1,3 +1,4 @@ +// Copyright (C) 2026-present The NetCalyx Authors. // Copyright (C) 2025-present The NetGauze Authors. // // Licensed under the Apache License, Version 2.0 (the "License"); @@ -162,7 +163,8 @@ mod tests { &mut exporter_peers, ); assert!(result1.is_none()); - // The buffer for this flow key should now contain the first part, so not empty + // The buffer for this flow key should now contain the first part, + // so not empty assert!(!exporter_peers.get(&flow_key).unwrap().1.is_empty()); // Second packet completes it diff --git a/crates/pcap-decoder/src/handlers/udp_notif.rs b/crates/pcap-decoder/src/handlers/udp_notif.rs index ce31e9c0..9cb78e71 100644 --- a/crates/pcap-decoder/src/handlers/udp_notif.rs +++ b/crates/pcap-decoder/src/handlers/udp_notif.rs @@ -51,8 +51,8 @@ impl ProtocolHandler .or_insert((UdpPacketCodec::default(), BytesMut::new())); buffer.extend_from_slice(packet_data); - // because of implementation specification UDP-Notif exports maximum 1 message - // per packet payload + // because of implementation specification UDP-Notif exports maximum + // 1 message per packet payload let mut results = Vec::new(); decode_buffer(buffer, codec, flow_key, &mut results); if !results.is_empty() { @@ -71,7 +71,8 @@ impl ProtocolHandler let (flow_key, udp_notif_packet) = m; let mut value = serde_json::to_value(&udp_notif_packet) .expect("Couldn't serialize UDP-Notif message to json"); - // Convert when possible inner payload into human-readable format + // Convert when possible inner payload into human-readable + // format match udp_notif_packet.media_type() { MediaType::YangDataJson => { let payload = serde_json::from_slice(&udp_notif_packet.payload()) @@ -168,10 +169,11 @@ mod tests { IpAddr::V4(Ipv4Addr::new(10, 0, 0, 2)), 4739, ); - // A datagram whose declared Message Length (14) exceeds the bytes actually - // present (12). UDP-Notif is datagram oriented: each datagram is complete on - // its own and must not be reassembled across datagrams, so a short datagram - // is malformed rather than an incomplete read awaiting more data. + // A datagram whose declared Message Length (14) exceeds the bytes + // actually present (12). UDP-Notif is datagram oriented: each datagram + // is complete on its own and must not be reassembled across datagrams, + // so a short datagram is malformed rather than an incomplete read + // awaiting more data. let truncated = [ 0x21, // version 1, no private space, Media type: 1 = YANG data JSON 0x0c, // Header length @@ -193,10 +195,12 @@ mod tests { UdpPacketCodecError::InvalidMessageLength(14) )]), ); - // The malformed bytes must be cleared, never retained for a later datagram. + // The malformed bytes must be cleared, never retained for a later + // datagram. assert!(exporter_peers.get(&flow_key).unwrap().1.is_empty()); - // A subsequent complete datagram decodes independently of the prior one. + // A subsequent complete datagram decodes independently of the prior + // one. let complete = [ 0x21, // version 1, no private space, Media type: 1 = YANG data JSON 0x0c, // Header length diff --git a/crates/pcap-reader/src/lib.rs b/crates/pcap-reader/src/lib.rs index f7f9d1b9..3bf4ac00 100644 --- a/crates/pcap-reader/src/lib.rs +++ b/crates/pcap-reader/src/lib.rs @@ -1,3 +1,4 @@ +// Copyright (C) 2026-present The NetCalyx Authors. // Copyright (C) 2023-present The NetGauze Authors. // // Licensed under the Apache License, Version 2.0 (the "License"); @@ -219,7 +220,8 @@ impl<'a> PcapIter<'a> { Ipv4::Udp(udp) => { let src_port = udp.source_port(); let dst_port = udp.destination_port(); - // UDP payload length, to avoiding parsing any padding bytes. + // UDP payload length, to avoiding parsing any padding + // bytes. let len = udp.length() as usize - 8; match udp.inner() { Err(_) => None, diff --git a/crates/udp-notif-pkt/src/codec.rs b/crates/udp-notif-pkt/src/codec.rs index ccf292fa..60308bdd 100644 --- a/crates/udp-notif-pkt/src/codec.rs +++ b/crates/udp-notif-pkt/src/codec.rs @@ -194,8 +194,8 @@ impl ReassemblyBuffer { // Per draft-ietf-netconf-udp-notif Section 4.1, all options (other than // the segmentation option) are carried on the first segment and are - // appended to the reassembled header. Options on subsequent segments are - // ignored. + // appended to the reassembled header. Options on subsequent segments + // are ignored. let options: HashMap<_, _> = first_segment .options() .iter() @@ -402,11 +402,11 @@ impl UdpPacketCodec { // Detect duplicate segment numbers: a segment whose (publisher_id, // message_id, seg_no) triple is already present in the buffer is a - // network retransmission and MUST be dropped (draft-ietf-netconf-udp-notif - // §4.1). Note: if a sender reuses a message_id before the old - // reassembly buffer has timed out, new segments will be dropped here - // too until the old buffer is evicted by the timeout. This is bounded - // by reassembly_timeout. + // network retransmission and MUST be dropped + // (draft-ietf-netconf-udp-notif §4.1). Note: if a sender reuses + // a message_id before the old reassembly buffer has timed out, + // new segments will be dropped here too until the old buffer is + // evicted by the timeout. This is bounded by reassembly_timeout. let is_duplicate = self .incomplete_messages .get(&message_key) @@ -418,8 +418,8 @@ impl UdpPacketCodec { return Ok(None); } - // Enforce the per-message segment cap (draft-ietf-netconf-udp-notif §5.2): - // reject if adding this segment would exceed the limit. + // Enforce the per-message segment cap (draft-ietf-netconf-udp-notif + // §5.2): reject if adding this segment would exceed the limit. if self .incomplete_messages .get(&message_key) @@ -466,7 +466,8 @@ impl Decoder for UdpPacketCodec { #[inline] fn decode(&mut self, buf: &mut BytesMut) -> Result, Self::Error> { // Evict reassembly buffers that have exceeded the configured timeout, - // accumulating the count into reassembly_events for the caller to report. + // accumulating the count into reassembly_events for the caller to + // report. let now = Instant::now(); let before = self.incomplete_messages.len(); self.incomplete_messages @@ -489,7 +490,8 @@ impl Decoder for UdpPacketCodec { let pkt_buf = buf.split_to(buf.len()); match UdpNotifPacket::from_wire(Span::new(pkt_buf.chunk())) { Ok((span, pkt)) => { - // Check that the message length matches the actual length of the message + // Check that the message length matches the actual length of + // the message if span.location_offset() != msg_len as usize || !span.is_empty() { return Err(UdpPacketCodecError::InvalidMessageLength(msg_len)); } @@ -827,7 +829,8 @@ mod tests { std::thread::sleep(Duration::from_millis(5)); - // An empty-buffer decode drives the eviction pass without consuming data. + // An empty-buffer decode drives the eviction pass without consuming + // data. let mut empty = BytesMut::new(); assert_eq!(codec.decode(&mut empty), Ok(None)); assert_eq!(codec.incomplete_messages_count(), 0); diff --git a/crates/udp-notif-pkt/src/decoded.rs b/crates/udp-notif-pkt/src/decoded.rs index bcb81043..045ca644 100644 --- a/crates/udp-notif-pkt/src/decoded.rs +++ b/crates/udp-notif-pkt/src/decoded.rs @@ -390,8 +390,8 @@ mod tests { let packet = UdpNotifPacket::new(MediaType::Unknown(99), 1234, 5678, HashMap::new(), payload); - // Attempt to decode the packet (will throw an error since the media type is - // unknown) + // Attempt to decode the packet + // (will throw an error since the media type is unknown) let result = UdpNotifPacketDecoded::try_from(&packet); assert!(matches!( result, @@ -406,8 +406,8 @@ mod tests { let packet = UdpNotifPacket::new(MediaType::YangDataJson, 1234, 5678, HashMap::new(), payload); - // Attempt to decode the packet (will throw an error since the payload is not - // valid JSON) + // Attempt to decode the packet + // (will throw an error since the payload is not valid JSON) let result = UdpNotifPacketDecoded::try_from(&packet); assert!( @@ -443,8 +443,8 @@ mod tests { Bytes::from(payload), ); - // Attempt to decode the packet (will throw an error since the - // NotificationVariant is unknown) + // Attempt to decode the packet + // (will throw an error since the NotificationVariant is unknown) let result = UdpNotifPacketDecoded::try_from(&packet); assert!( diff --git a/crates/udp-notif-pkt/src/notification.rs b/crates/udp-notif-pkt/src/notification.rs index b2e61cc8..e400e46c 100644 --- a/crates/udp-notif-pkt/src/notification.rs +++ b/crates/udp-notif-pkt/src/notification.rs @@ -1073,7 +1073,8 @@ mod tests { let deserialized_envelope: NotificationEnvelope = serde_json::from_value(serialized_envelope).expect("Deserialization failed"); - // Assert that the deserialized NotificationEnvelope matches the original + // Assert that the deserialized NotificationEnvelope matches + // the original assert_eq!(notification_envelope, deserialized_envelope); } @@ -1271,7 +1272,8 @@ mod tests { let deserialized_envelope: NotificationEnvelope = serde_json::from_value(serialized_envelope).expect("Deserialization failed"); - // Assert that the deserialized NotificationEnvelope matches the original + // Assert that the deserialized NotificationEnvelope matches + // the original assert_eq!(notification_envelope, deserialized_envelope); } @@ -1437,7 +1439,8 @@ mod tests { let deserialized_envelope: NotificationEnvelope = serde_json::from_value(serialized_envelope).expect("Deserialization failed"); - // Assert that the deserialized NotificationEnvelope matches the original + // Assert that the deserialized NotificationEnvelope matches + // the original assert_eq!(notification_envelope, deserialized_envelope); } diff --git a/crates/udp-notif-pkt/src/wire/deserialize.rs b/crates/udp-notif-pkt/src/wire/deserialize.rs index b4f1a28c..f87eeddc 100644 --- a/crates/udp-notif-pkt/src/wire/deserialize.rs +++ b/crates/udp-notif-pkt/src/wire/deserialize.rs @@ -1,3 +1,4 @@ +// Copyright (C) 2026-present The NetCalyx Authors. // Copyright (C) 2024-present The NetGauze Authors. // // Licensed under the Apache License, Version 2.0 (the "License"); @@ -51,7 +52,8 @@ impl<'a> ReadablePdu<'a, LocatedUdpNotifOptionParsingError<'a>> for UdpNotifOpti let (value_buf, high) = nom::number::complete::u8(value_buf)?; let (_value_buf, low) = nom::number::complete::u8(value_buf)?; let number = ((high as u16) << 7) | ((low as u16) >> 1); - // Extract the L flag (the least significant bit of the last byte) + // Extract the L flag (the least significant bit of the last + // byte) let last = (low & 0x01) != 0; Ok((buf, UdpNotifOption::Segment { number, last })) @@ -150,9 +152,10 @@ impl<'a> ReadablePdu<'a, LocatedUdpNotifPacketParsingError<'a>> for UdpNotifPack let (header_buf, publisher_id) = be_u32(header_buf)?; let (mut header_buf, message_id) = be_u32(header_buf)?; let mut options = HashMap::new(); - // AS per UDP NOTIF RFC: When S is set, MT represents a private space to be - // freely used for non standard encodings. When S is set, the Private - // Encoding Option SHOULD be present in the UDP-Notif message header. + // AS per UDP NOTIF RFC: When S is set, MT represents a private space + // to be freely used for non standard encodings. When S is set, + // the Private Encoding Option SHOULD be present in the UDP-Notif + // message header. let mut private_is_correct = !s_flag; while !header_buf.is_empty() { let (t, option) = parse_into_located::< diff --git a/crates/udp-notif-pkt/src/wire/serialize.rs b/crates/udp-notif-pkt/src/wire/serialize.rs index ad2c7488..8aa6ec29 100644 --- a/crates/udp-notif-pkt/src/wire/serialize.rs +++ b/crates/udp-notif-pkt/src/wire/serialize.rs @@ -1,3 +1,4 @@ +// Copyright (C) 2026-present The NetCalyx Authors. // Copyright (C) 2024-present The NetGauze Authors. // // Licensed under the Apache License, Version 2.0 (the "License"); @@ -43,8 +44,8 @@ impl WritablePdu for UdpNotifOption { fn len(&self) -> usize { match self { UdpNotifOption::Segment { .. } => { - // base length + two octets for segment length of which the last bit is a `last - // segment` flag + // base length + two octets for segment length of which the last + // bit is a `last segment` flag Self::BASE_LENGTH + 2 } UdpNotifOption::PrivateEncoding(value) => Self::BASE_LENGTH + value.len(), diff --git a/crates/udp-notif-pkt/src/wire/test/pcap_tests.rs b/crates/udp-notif-pkt/src/wire/test/pcap_tests.rs index a1b5f515..205c2f7e 100644 --- a/crates/udp-notif-pkt/src/wire/test/pcap_tests.rs +++ b/crates/udp-notif-pkt/src/wire/test/pcap_tests.rs @@ -99,8 +99,8 @@ fn test_udp_notif_pcap(overwrite: bool, pcap_path: PathBuf) { let iter = PcapIter::new(Box::new(pcap_reader)); let mut peers = HashMap::new(); for (src_ip, src_port, dst_ip, dst_port, protocol, value) in iter { - // The filter for 161 is included because n7-sa1_yang-push.pcap have some snmp - // traffic + // The filter for 161 is included because n7-sa1_yang-push.pcap have + // some snmp traffic if protocol != TransportProtocol::UDP || ![10003, 10100, 57499].contains(&dst_port) || src_port == 161 @@ -118,7 +118,8 @@ fn test_udp_notif_pcap(overwrite: bool, pcap_path: PathBuf) { Ok(Some(msg)) => { let mut udp_notif_value = serde_json::to_value(&msg) .expect("Couldn't serialize UDP-Notif message to json"); - // Convert when possible inner payload into human-readable format + // Convert when possible inner payload into human-readable + // format let decoded = UdpNotifPacketDecoded::try_from(&msg) .map_err(|err| format!("Couldn't decode UDP-Notif message: {err}")); match msg.media_type() { @@ -179,8 +180,8 @@ fn test_udp_notif_pcap(overwrite: bool, pcap_path: PathBuf) { (udp_notif_value, decoded) } Ok(None) => { - // packet is fragmented, need to read the next PDU first before attempting to - // deserialize it + // packet is fragmented, need to read the next PDU first + // before attempting to deserialize it break; } Err(err) => { @@ -217,7 +218,8 @@ fn test_udp_notif_pcap(overwrite: bool, pcap_path: PathBuf) { ); let expected = lines.next().expect(&err_msg).expect("Error reading"); - // Compare JSON values to succeed even when key order in string values differs + // Compare JSON values to succeed even when key order in string + // values differs let expected_json: Value = serde_json::from_str(&expected).expect("Failed to parse expected JSON"); diff --git a/crates/udp-notif-service/examples/udp-notif-print-decoded.rs b/crates/udp-notif-service/examples/udp-notif-print-decoded.rs index 4a34aa93..1149c858 100644 --- a/crates/udp-notif-service/examples/udp-notif-print-decoded.rs +++ b/crates/udp-notif-service/examples/udp-notif-print-decoded.rs @@ -74,16 +74,19 @@ async fn main() -> Result<(), Box while let Some(next) = stream.next().await { match next { Ok((mut buf, addr)) => { - // Peek at segment metadata — we'll use this after draining codec events - // so that eviction cleanup runs before we increment the pending counter. + // Peek at segment metadata — we'll use this after draining + // codec events so that eviction cleanup runs + // before we increment the pending counter. let seg_info = peek_segment_info(&buf); - // If we haven't seen the client before, create a new UdpPacketCodec for it. - // UdpPacketCodec handles the decoding/encoding of udp-notif packets. + // If we haven't seen the client before, create a new + // UdpPacketCodec for it. UdpPacketCodec handles + // the decoding/encoding of udp-notif packets. let codec = clients.entry(addr).or_default(); let result = codec.decode(&mut buf); - // Drain reassembly event counts and log any anomalies with peer context. + // Drain reassembly event counts and log any anomalies with peer + // context. let reassembly_events = codec.take_reassembly_events(); if reassembly_events.timeout_evictions > 0 { warn!( @@ -91,8 +94,9 @@ async fn main() -> Result<(), Box evicted = reassembly_events.timeout_evictions, "evicted timed-out reassembly buffers" ); - // We don't know which (publisher_id, message_id) keys were evicted, - // so clear all pending state for this peer to avoid stale counts. + // We don't know which (publisher_id, message_id) keys + // were evicted, so clear all pending state for this peer + // to avoid stale counts. pending.remove(&addr); } if reassembly_events.duplicate_drops > 0 { @@ -103,8 +107,8 @@ async fn main() -> Result<(), Box ); } - // Now that stale pending state has been cleared, update the counter - // for the segment that was just processed. + // Now that stale pending state has been cleared, update + // the counter for the segment that was just processed. if let Some((pub_id, msg_id, seg_no, is_last)) = seg_info { let received = pending .entry(addr) @@ -127,10 +131,12 @@ async fn main() -> Result<(), Box Ok(Some(msg)) => { let pub_id = msg.publisher_id(); let msg_id = msg.message_id(); - // Always clean up the pending counter using the message's own IDs, - // regardless of seg_info. This handles the case where the - // reassembly-triggering segment had no Segment option (e.g. a - // single-segment retransmission after a prior segmented run timed out). + // Always clean up the pending counter using the + // message's own IDs, regardless of seg_info. This + // handles the case where the reassembly-triggering + // segment had no Segment option (e.g. a single-segment + // retransmission after a prior segmented run timed + // out). if let Some(total) = pending .get_mut(&addr) .and_then(|m| m.remove(&(pub_id, msg_id))) diff --git a/crates/udp-notif-service/examples/udp-notif-print.rs b/crates/udp-notif-service/examples/udp-notif-print.rs index 82530d22..62d75ab5 100644 --- a/crates/udp-notif-service/examples/udp-notif-print.rs +++ b/crates/udp-notif-service/examples/udp-notif-print.rs @@ -47,8 +47,9 @@ async fn main() -> Result<(), Box while let Some(next) = stream.next().await { match next { Ok((mut buf, addr)) => { - // If we haven't seen the client before, create a new UdpPacketCodec for it. - // UdpPacketCodec handles the decoding/encoding of udp-notif packets. + // If we haven't seen the client before, create a new + // UdpPacketCodec for it. UdpPacketCodec handles the + // decoding/encoding of udp-notif packets. let result = clients .entry(addr) .or_insert(UdpPacketCodec::default()) diff --git a/crates/udp-notif-service/src/actor.rs b/crates/udp-notif-service/src/actor.rs index 788ebcf5..4afc3898 100644 --- a/crates/udp-notif-service/src/actor.rs +++ b/crates/udp-notif-service/src/actor.rs @@ -314,8 +314,8 @@ impl UdpNotifActor { next: (BytesMut, SocketAddr), ) -> Option<(SocketAddr, UdpNotifPacket)> { let (mut buf, addr) = next; - // If we haven't seen the client before, create a new UdpPacketCodec for it. - // UdpPacketCodec handles the decoding/encoding of packets. + // If we haven't seen the client before, create a new UdpPacketCodec for + // it. UdpPacketCodec handles the decoding/encoding of packets. let codec = self.clients.entry(addr).or_insert_with(|| { UdpPacketCodec::new(self.reassembly_max_segments, self.reassembly_timeout) }); @@ -443,8 +443,8 @@ impl UdpNotifActor { sent_counter: opentelemetry::metrics::Counter, drop_counter: opentelemetry::metrics::Counter, ) { - // The send operation is bounded by timeout period to avoid blocking on a slow - // subscriber. + // The send operation is bounded by timeout period to avoid blocking on + // a slow subscriber. let ref_clone = msg.clone(); let drop_counter_clone = drop_counter.clone(); let peer = msg.peer_address(); @@ -913,7 +913,8 @@ impl UdpNotifActor { .map_err(|err| { UdpNotifActorError::SocketBindError(self.actor_id, self.socket_addr, err) })?; - // Get the local address of the socket, handy in cases where the port is 0 + // Get the local address of the socket, handy in cases where the port is + // 0 self.socket_addr = socket.local_addr().map_err(|err| { UdpNotifActorError::GetLocalAddressError(self.actor_id, self.socket_addr, err) })?; @@ -1126,8 +1127,8 @@ impl ActorHandle { duration: Duration, ) -> Result, ActorHandleError> { let (tx, mut rx) = mpsc::channel(self.cmd_buffer_size); - // If the command fails, the recv after will fail, no need to double handle the - // error + // If the command fails, the recv after will fail, + // no need to double handle the error self.cmd_tx .send(ActorCommand::PurgeUnusedPeers(duration, tx)) .await diff --git a/crates/udp-notif-service/src/supervisor.rs b/crates/udp-notif-service/src/supervisor.rs index 1ff3822f..1bb2688b 100644 --- a/crates/udp-notif-service/src/supervisor.rs +++ b/crates/udp-notif-service/src/supervisor.rs @@ -144,8 +144,8 @@ impl UdpNotifSupervisor { let awaited = futures::future::select_all(handles); handles = match awaited.await { (Ok(ret), _, rest) => { - // TODO(AH): Have some policy to allow to restart actors or terminate supervisor - // if failed + // TODO(AH): Have some policy to allow to restart actors + // or terminate supervisor if failed if let Err(err) = ret { error!("[Supervisor] actor terminated with error: {}", err); } diff --git a/crates/yang-push/src/cache/actor.rs b/crates/yang-push/src/cache/actor.rs index 2e77beac..4d3ba9d6 100644 --- a/crates/yang-push/src/cache/actor.rs +++ b/crates/yang-push/src/cache/actor.rs @@ -783,8 +783,8 @@ impl CacheActor { None } }; - // First, remove all pending requests for this subscription info that are - // requested with full subscription info + // First, remove all pending requests for this subscription info + // that are requested with full subscription info let empty = SubscriptionInfo::new_empty( subscription_info.peer_ip(), subscription_info.id(), @@ -874,7 +874,8 @@ impl CacheActor { } None => { // YANG Library Reference is not found in the cache. - // Start a new worker to fetch the YANG Library from the server. + // Start a new worker to fetch the YANG Library from the + // server. self.stats.cache_misses.add(1, &otl_tags); let entry = self .pending_requests @@ -909,8 +910,8 @@ impl CacheActor { error=%err, "failed to fetch yang library from device" ); - // remove the sender we just added since the fetch failed to - // start + // remove the sender we just added + // since the fetch failed to start entry.remove(entry.len() - 1); self.stats.device_fetch_failed.add(1, &otl_tags); return; @@ -1059,8 +1060,8 @@ impl CacheActor { error=%err, "failed to fetch yang library from device" ); - // remove the sender we just added since the fetch failed to - // start + // remove the sender we just added + // since the fetch failed to start entry.remove(entry.len() - 1); self.stats.device_fetch_failed.add(1, &otel_tags); return; diff --git a/crates/yang-push/src/cache/fetcher.rs b/crates/yang-push/src/cache/fetcher.rs index 18504975..7cbc50a6 100644 --- a/crates/yang-push/src/cache/fetcher.rs +++ b/crates/yang-push/src/cache/fetcher.rs @@ -378,8 +378,8 @@ impl NetconfYangLibraryFetcher { Arc::clone(&cfg.client_config), ) .with_module_cache(cfg.module_cache.clone()); - // Empty subscription info returned in case of errors to keep track of peer and - // subscription ID + // Empty subscription info returned in case of errors to keep track of + // peer and subscription ID let empty = SubscriptionInfo::new_empty(peer_ip, subscription_id); let mut client = match tokio::time::timeout(cfg.timeout, connect(config)).await { Ok(Ok(c)) => c, @@ -415,7 +415,8 @@ impl NetconfYangLibraryFetcher { Target::Stream(stream_target) => match &stream_target.filter { StreamSelectionFilterObjects::ByReference(name) => { // references are resolved in the NETCONF client, - // if we reach this point, there must be a misconfigured router, + // if we reach this point, there must be a misconfigured + // router error!( %name, subscription_id, diff --git a/crates/yang-push/src/cache/storage.rs b/crates/yang-push/src/cache/storage.rs index 5c0331bd..c48253a5 100644 --- a/crates/yang-push/src/cache/storage.rs +++ b/crates/yang-push/src/cache/storage.rs @@ -1143,14 +1143,15 @@ impl YangLibraryCache { } } - // Step 3: Check if any other subscriptions are still using this content_id + // Step 3: Check if any other subscriptions are still using this + // content_id let content_id_still_in_use = self .cache_by_subscription_info .values() .any(|lib_ref| lib_ref.content_id() == yang_lib_ref.content_id()); - // Step 4: Only remove YangLibraryReference and delete files if no subscriptions - // remain + // Step 4: Only remove YangLibraryReference and delete files + // if no subscriptions remain if !content_id_still_in_use { if let Some(yang_lib_ref) = self.cache_by_content_id.remove(yang_lib_ref.content_id()) { // Remove subscription from the reference's internal list @@ -1168,8 +1169,8 @@ impl YangLibraryCache { } } } else { - // Step 5: Content is still in use - just remove this subscription from the - // reference + // Step 5: Content is still in use - just remove this subscription + // from the reference if let Some(yang_lib_ref) = self.cache_by_content_id.get_mut(yang_lib_ref.content_id()) { yang_lib_ref.remove_subscription_info(subscription_info)?; @@ -1930,8 +1931,8 @@ mod tests { assert_eq!(cache.cache_by_subscription_info.len(), 1); assert_eq!(cache.cache_by_subscription_id.len(), 1); - // Create second subscription info with different peer and target but same - // content_id + // Create second subscription info with different peer and target but + // same content_id let subscription_info2 = SubscriptionInfo::new( IpAddr::from([192, 168, 1, 102]), 2, diff --git a/crates/yang-push/src/model/telemetry.rs b/crates/yang-push/src/model/telemetry.rs index 85c1769b..a81313ac 100644 --- a/crates/yang-push/src/model/telemetry.rs +++ b/crates/yang-push/src/model/telemetry.rs @@ -700,7 +700,8 @@ mod tests { // Expected JSON string let expected_json = r#"{"ietf-telemetry-message:message":{"network-node-manifest":{"name":"node_id","vendor":"FRR"},"telemetry-message-metadata":{"collection-timestamp":"1970-01-01T00:00:00Z","notification-event":"log","sequence-number":1,"session-protocol":"yang-push","export-address":"127.0.0.1","export-port":8080,"ietf-yang-push-telemetry-message:yang-push-subscription":{"id":1,"stream":"example-stream-subtree-filter-map","subtree-filter":{"example-map":{"e1":"v1","e2":"v2"}},"transport":"ietf-udp-notif-transport:udp-notif","encoding":"ietf-subscribed-notifications:encode-json","periodic":{"period":100,"anchor-time":"1970-01-01T00:00:00Z"},"module":[{"name":"example-module","revision":"2025-01-01","version":"1.0.0"}],"yang-library-content-id":"random-content-id"}},"data-collection-manifest":{"name":"dev-collector","vendor":"NetCalyx","vendor-pen":12345,"software-version":"1.0.0","software-flavor":"release","os-version":"8.10","os-type":"Rocky Linux"},"network-operator-metadata":{"labels":[{"name":"priority_level","number-value":"100"},{"name":"platform_id","string-value":"IETF LAB"},{"name":"test_anykey_label","anydata-values":{"key":"value"}}]}}}"#; - // Assert that the serialized JSON string matches the expected JSON string + // Assert that the serialized JSON string matches the expected + // JSON string assert_eq!( serialized, expected_json, "Serialized JSON does not match the expected JSON" @@ -743,7 +744,8 @@ mod tests { // Expected JSON string let expected_json = r#"{"ietf-telemetry-message:message":{"telemetry-message-metadata":{"collection-timestamp":"1970-01-01T00:00:00Z","notification-event":"log","session-protocol":"unknown","export-address":"127.0.0.1"}}}"#; - // Assert that the serialized JSON string matches the expected JSON string + // Assert that the serialized JSON string matches the expected + // JSON string assert_eq!( serialized, expected_json, "Serialized JSON does not match the expected JSON" diff --git a/crates/yang-push/src/validation/mod.rs b/crates/yang-push/src/validation/mod.rs index 9be2bc82..02f65787 100644 --- a/crates/yang-push/src/validation/mod.rs +++ b/crates/yang-push/src/validation/mod.rs @@ -689,7 +689,8 @@ impl ValidationActor { let publisher_id = packet.publisher_id(); let mut peer_tags = Self::peer_tags_from_packet(peer, packet); - // Decode the UDP-Notif packet to get subscription ID and payload information + // Decode the UDP-Notif packet to get subscription ID and payload + // information match UdpNotifPacketDecoded::try_from(packet) { Ok(decoded) => { let notif_contents = decoded.payload().notification_contents(); @@ -1004,8 +1005,9 @@ impl ValidationActor { .map(|x| x.to_string()) .unwrap_or("UNKNOWN".to_string()); - // Defer started/modified notifications while a fetch is already in-flight - // for this id, so a stale response can never clobber a newer generation. + // Defer started/modified notifications while a fetch is already + // in-flight for this id, so a stale response can never clobber + // a newer generation. if let Some(notif_contents) = decoded.payload().notification_contents() && matches!( notif_contents, @@ -1052,12 +1054,15 @@ impl ValidationActor { return Ok(Some(subscription_info)); } Some(None) => { - // Cache entry exists but schema not yet available. We distinguish: - // - fetch in-flight (schema_fetch_pending = true): buffer the packet so it - // is validated once the response arrives, instead of slipping through + // Cache entry exists but schema not yet available. We + // distinguish: + // - fetch in-flight (schema_fetch_pending = true): + // buffer the packet so it is validated once the + // response arrives, instead of slipping through // unvalidated. - // - fetch already completed with no schema (schema_fetch_pending = false): - // forward unvalidated as usual; no point buffering. + // - fetch already completed with no schema + // (schema_fetch_pending = false): forward unvalidated + // as usual; no point buffering. let fetch_pending = self .peer_cache .get(&peer.ip()) @@ -1137,10 +1142,10 @@ impl ValidationActor { Some(NotificationVariant::SubscriptionStarted(_)) | Some(NotificationVariant::SubscriptionModified(_)) ) { - // A subscription-started/modified that reached here failed to - // build SubscriptionInfo (e.g. missing module version). It will - // fail identically every time, so buffering it and re-fetching - // would loop forever. Drop it permanently. + // A subscription-started/modified that reached here failed + // to build SubscriptionInfo (e.g. missing module version). + // It will fail identically every time, so buffering it and + // re-fetching would loop forever. Drop it permanently. warn!( peer=%peer, message_id, @@ -1302,7 +1307,8 @@ impl ValidationActor { subscription_cache.schema_fetch_pending = false; let buffered_packets = std::mem::take(&mut subscription_cache.buffered_packets); let drained = buffered_packets.len(); - // Update the per-peer counter while we still hold the peer_cache borrow. + // Update the per-peer counter while we still hold the peer_cache + // borrow. peer_cache.total_buffered -= drained; let remaining = peer_cache.total_buffered; for message in buffered_packets { @@ -1995,7 +2001,8 @@ mod tests { .await .unwrap(); - // Nothing should be forwarded; packet is dropped, no fetch is triggered. + // Nothing should be forwarded; packet is dropped, + // no fetch is triggered. let res = tokio::time::timeout(Duration::from_millis(300), validated_rx.recv()).await; assert!(res.is_err(), "malformed packet must not be forwarded"); assert!( @@ -2165,7 +2172,8 @@ mod tests { handle, ) = setup_validation_actor(); - // YangDataJson with bytes that are not valid JSON → serde_json parse error. + // YangDataJson with bytes that are not valid JSON → serde_json parse + // error. udp_notif_tx .send(Arc::new(UdpNotifRequest::new( SessionInfo::new( @@ -2214,8 +2222,8 @@ mod tests { let peer = SocketAddr::new(subscription_info.peer_ip(), 0); setup_and_load_schema(&udp_notif_tx, &validated_rx, peer).await; - // Push-update with "enabelled" (typo for "enabled"): an unknown YANG node - // that strict validation must reject. + // Push-update with "enabelled" (typo for "enabled"): + // an unknown YANG node that strict validation must reject. let invalid_push_update_payload = serde_json::json!({ "ietf-yp-notification:envelope": { "event-time": "2026-04-21T13:33:31.007Z", @@ -2299,8 +2307,8 @@ mod tests { let peer = SocketAddr::new(subscription_info.peer_ip(), 0); setup_and_load_schema(&udp_notif_tx, &validated_rx, peer).await; - // Push-update with "enabelled" (typo for "enabled"): an unknown YANG node - // that only strict anydata validation would reject. + // Push-update with "enabelled" (typo for "enabled"): an unknown YANG + // node that only strict anydata validation would reject. let invalid_push_update_payload = serde_json::json!({ "ietf-yp-notification:envelope": { "event-time": "2026-04-21T13:33:31.007Z", @@ -2429,8 +2437,8 @@ mod tests { .await .unwrap(); - // TODO(libyang): should be `res.is_err()` once libyang enforces mandatory nodes - // inside anydata. + // TODO(libyang): should be `res.is_err()` once libyang enforces + // mandatory nodes inside anydata. let res = tokio::time::timeout(Duration::from_millis(300), validated_rx.recv()).await; assert!( res.is_ok(), @@ -2517,11 +2525,13 @@ mod tests { // Send first SubscriptionStarted. This triggers the schema fetch and // sets schema_fetch_pending = true. udp_notif_tx.send(make_packet(1)).await.unwrap(); - // Yield so the actor processes the first packet before the duplicate is queued. + // Yield so the actor processes the first packet before the duplicate is + // queued. tokio::task::yield_now().await; - // Send the duplicate while the fetch is in-flight. With schema_fetch_pending = - // true the actor must buffer it rather than forwarding it unvalidated. + // Send the duplicate while the fetch is in-flight. With + // schema_fetch_pending = true the actor must buffer it + // rather than forwarding it unvalidated. udp_notif_tx.send(make_packet(2)).await.unwrap(); // Both packets must eventually be forwarded with a valid content_id. @@ -2543,7 +2553,8 @@ mod tests { assert!(!sub_info.is_empty()); } - // Exactly one cache fetch must have been triggered for both identical packets. + // Exactly one cache fetch must have been triggered for both identical + // packets. assert_eq!( fetcher_count.lock().unwrap().len(), 1, @@ -2572,7 +2583,8 @@ mod tests { ) = setup_validation_actor(); let peer = SocketAddr::new(subscription_info.peer_ip(), 0); - // Load the schema via the first SubscriptionStarted and drain the result. + // Load the schema via the first SubscriptionStarted and drain the + // result. setup_and_load_schema(&udp_notif_tx, &validated_rx, peer).await; assert_eq!( fetcher_count.lock().unwrap().len(), @@ -2744,8 +2756,9 @@ mod tests { "forwarded SessionInfo must retain the new source port, not the original one" ); - // No additional cache fetch must have been triggered: the peer/subscription - // must be recognized from the IP alone, regardless of source port. + // No additional cache fetch must have been triggered: + // the peer/subscription must be recognized from the IP alone, + // regardless of source port. assert_eq!( fetcher_count.lock().unwrap().len(), 1, @@ -2836,11 +2849,11 @@ mod tests { .await .unwrap(); - // The test fetcher only knows "test-content-id-1" and returns an error for - // "updated-content-id-2" (simulating a failed device fetch). The packet was - // buffered during the fetch attempt; after the fetch fails it is forwarded - // unvalidated. In production the fetch would succeed and content_id would be - // Some. + // The test fetcher only knows "test-content-id-1" and returns an error + // for "updated-content-id-2" (simulating a failed device fetch). The + // packet was buffered during the fetch attempt; after the fetch fails + // it is forwarded unvalidated. In production the fetch would succeed + // and content_id would be Some. let ValidatedNotification { cached_content_id: content_id, subscription_info: sub_info, @@ -2855,8 +2868,9 @@ mod tests { ); assert!(!sub_info.is_empty()); - // A second device fetch must have been triggered for the new content-id; - // the cache must not silently reuse the old schema when content-id changes. + // A second device fetch must have been triggered for the new + // content-id; the cache must not silently reuse the old schema + // when content-id changes. assert_eq!( fetcher_count.lock().unwrap().len(), 2, @@ -3003,7 +3017,8 @@ mod tests { .await .unwrap(); - // Messages 1 and 2 must be validated with the first (content-id-1) schema. + // Messages 1 and 2 must be validated with the first (content-id-1) + // schema. for expected_msg_id in [1u32, 2u32] { let notification = tokio::time::timeout(Duration::from_secs(3), validated_rx.recv()) .await @@ -3017,8 +3032,9 @@ mod tests { ); } - // Messages 3 (deferred SubscriptionModified) and 4 must come after, once the - // second fetch (for content-id-2, unknown to the test fetcher) fails. + // Messages 3 (deferred SubscriptionModified) and 4 must come after, + // once the second fetch (for content-id-2, unknown to the test fetcher) + // fails. for expected_msg_id in [3u32, 4u32] { let notification = tokio::time::timeout(Duration::from_secs(3), validated_rx.recv()) .await diff --git a/fuzz/fuzz_targets/fuzz_bgp_peer.rs b/fuzz/fuzz_targets/fuzz_bgp_peer.rs index 60559fbf..caf95dc8 100644 --- a/fuzz/fuzz_targets/fuzz_bgp_peer.rs +++ b/fuzz/fuzz_targets/fuzz_bgp_peer.rs @@ -1,3 +1,4 @@ +// Copyright (C) 2026-present The NetCalyx Authors. // Copyright (C) 2024-present The NetGauze Authors. // // Licensed under the Apache License, Version 2.0 (the "License"); @@ -443,8 +444,8 @@ fuzz_target!( if peer.fsm_state() == FsmState::Idle || peer.fsm_state() == FsmState::Active { - // Peer with terminated or reached an active state in which it waits for - // TCP connection + // Peer with terminated or reached an active state + // in which it waits for TCP connection return; } }