Skip to content

feat(yang-push): make anydata_strict validation configurable #42

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
rodonile wants to merge 1 commit into
base: main
Choose a base branch
from
Open

feat(yang-push): make anydata_strict validation configurable #42

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 3 additions & 0 deletions crates/collector/config_udpnotif_telemetry.yaml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,9 @@ udp_notif:

cache_location: /tmp/netcalyx-udpnotif-telemetry-cache

# Strictly validate anydata nodes against their schema
anydata_strict_validation: true

netconf:
username: someusername
private_key_path: /paht/to/private-key
Expand Down
9 changes: 9 additions & 0 deletions crates/collector/src/config.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -71,6 +71,10 @@ pub(crate) const fn default_max_cached_packets_per_subscription() -> usize {
100
}

pub(crate) const fn default_anydata_strict_validation() -> bool {
true
}

pub(crate) const fn default_netconf_port() -> u16 {
830
}
Expand Down Expand Up @@ -250,6 +254,11 @@ pub struct UdpNotifConfig {
#[serde(default = "default_max_cached_packets_per_subscription")]
pub max_cached_packets_per_subscription: usize,

/// Whether `anydata` nodes are validated strictly against the
/// schema when validating YANG-Push notifications
#[serde(default = "default_anydata_strict_validation")]
pub anydata_strict_validation: bool,

pub netconf: NetconfConfig,

pub publishers: HashMap<String, PublisherConfig>,
Expand Down
2 changes: 2 additions & 0 deletions crates/collector/src/lib.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -619,6 +619,7 @@ pub async fn init_udp_notif_collection(
publisher_config.buffer_size,
udp_notif_config.max_cached_packets_per_peer,
udp_notif_config.max_cached_packets_per_subscription,
udp_notif_config.anydata_strict_validation,
udp_notif_recv.clone(),
validated_tx,
schema_handle.request_tx(),
Expand Down Expand Up @@ -694,6 +695,7 @@ pub async fn init_udp_notif_collection(
publisher_config.buffer_size,
udp_notif_config.max_cached_packets_per_peer,
udp_notif_config.max_cached_packets_per_subscription,
udp_notif_config.anydata_strict_validation,
udp_notif_recv.clone(),
validated_tx,
schema_handle.request_tx(),
Expand Down
110 changes: 109 additions & 1 deletion crates/yang-push/src/validation/mod.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -32,6 +32,7 @@
//! 100, // cache response channel buffer size
//! 1000, // max packets buffered per peer
//! 100, // max packets buffered per subscription
//! true, // enforce strict validation of anydata nodes
//! rx, // incoming UDP-Notif packets
//! tx, // validated packets output
//! cache_cmd_tx, // cache lookup commands
Expand Down Expand Up @@ -424,6 +425,9 @@ pub struct ValidatedNotification {
struct ValidationActor {
max_buffered_packets_per_peer: usize,
max_buffered_packets_per_subscription: usize,
/// Whether `anydata` nodes are validated strictly against their
/// schema (see `yang5::data::DataParserFlags::ANYDATA_STRICT`).
anydata_strict: bool,
peer_cache: FxHashMap<IpAddr, CachedPeerSubscriptions>,
cmd_rx: mpsc::Receiver<ValidationActorCommand>,
rx: async_channel::Receiver<Arc<UdpNotifRequest>>,
Expand Down Expand Up @@ -791,6 +795,7 @@ impl ValidationActor {
&notification_type,
yang_ctx,
is_legacy,
self.anydata_strict,
&self.stats,
&peer_tags,
);
Expand Down Expand Up @@ -883,18 +888,23 @@ impl ValidationActor {
notification_type: &String,
yang_ctx: &yang5::context::Context,
is_legacy: bool,
anydata_strict: bool,
stats: &ValidationStats,
peer_tags: &[opentelemetry::KeyValue],
) -> Result<(), yang5::Error> {
let message_id = packet.message_id();
let publisher_id = packet.publisher_id();

if !is_legacy {
let mut parser_flags = DataParserFlags::STRICT;
if anydata_strict {
parser_flags |= DataParserFlags::ANYDATA_STRICT;
}
let validation_result = yang5::data::DataTree::parse_string(
Comment thread
rodonile marked this conversation as resolved.
yang_ctx,
packet.payload(),
DataFormat::JSON,
DataParserFlags::STRICT | DataParserFlags::ANYDATA_STRICT,
parser_flags,
DataValidationFlags::PRESENT,
);
if let Err(err) = validation_result {
Expand Down Expand Up @@ -1473,10 +1483,12 @@ pub struct ValidationActorHandle {
}

impl ValidationActorHandle {
#[allow(clippy::too_many_arguments)]
pub fn new(
buffer_size: usize,
max_buffered_packets_per_peer: usize,
max_buffered_packets_per_subscription: usize,
anydata_strict: bool,
Comment thread
rodonile marked this conversation as resolved.
rx: async_channel::Receiver<Arc<UdpNotifRequest>>,
tx: async_channel::Sender<ValidatedNotification>,
cache_cmd_tx: async_channel::Sender<CacheLookupCommand>,
Expand All @@ -1497,6 +1509,7 @@ impl ValidationActorHandle {
let actor = ValidationActor {
max_buffered_packets_per_peer,
max_buffered_packets_per_subscription,
anydata_strict,
peer_cache: FxHashMap::default(),
cmd_rx,
rx,
Expand Down Expand Up @@ -1551,6 +1564,7 @@ mod tests {
100,
1000,
100,
true,
udp_notif_rx,
validated_tx,
caching_handle.request_tx(),
Expand Down Expand Up @@ -1847,6 +1861,7 @@ mod tests {
100,
10000,
1000,
true,
udp_notif_rx,
validated_tx,
caching_handle.request_tx(),
Expand Down Expand Up @@ -2256,6 +2271,99 @@ mod tests {
caching_join_handle.await.unwrap().unwrap();
}

/// Same push-update as `test_validation_actor_invalid_push_update_dropped`,
/// but with `anydata_strict` disabled: content of anydata nodes is not
/// being validated so the message must pass through
#[tokio::test]
#[tracing_test::traced_test]
async fn test_validation_actor_invalid_push_update_passes_when_anydata_not_strict() {
let (caching_join_handle, caching_handle, subscription_info, _fetcher_count) =
setup_actor_with_empty_cache();

let (udp_notif_tx, udp_notif_rx) = async_channel::bounded(100);
let (validated_tx, validated_rx) = async_channel::bounded(100);
let (_join_handle, handle) = ValidationActorHandle::new(
100,
1000,
100,
false, // anydata_strict disabled
udp_notif_rx,
validated_tx,
caching_handle.request_tx(),
either::Right(ValidationStats::new(opentelemetry::global::meter(
"test_meter",
))),
)
.expect("Failed to spawn validation actor");

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.
let invalid_push_update_payload = serde_json::json!({
"ietf-yp-notification:envelope": {
"event-time": "2026-04-21T13:33:31.007Z",
"hostname": "test-router-01",
"sequence-number": 1,
"contents": {
"ietf-yang-push:push-update": {
"id": 1,
"datastore-contents": {
"ietf-interfaces:interfaces": {
"interface": [
{
"name": "GigabitEthernet0/0/0",
"type": "iana-if-type:ethernetCsmacd",
"enabelled": true,
"admin-status": "up",
"oper-status": "up",
"if-index": 1,
"speed": "1000000000"
}
]
}
},
"ietf-distributed-notif:message-publisher-id": 16974839
}
}
}
});
let bytes = serde_json::to_vec(&invalid_push_update_payload).unwrap();
udp_notif_tx
.send(Arc::new(UdpNotifRequest::new(
SessionInfo::new(SocketAddr::from(([127, 0, 0, 1], 10000)), None, peer),
UdpNotifPacket::new(
MediaType::YangDataJson,
10,
2,
HashMap::new(),
Bytes::from(bytes),
),
)))
.await
.unwrap();

let ValidatedNotification {
cached_content_id: content_id,
subscription_info: sub_info,
..
} = tokio::time::timeout(Duration::from_secs(1), validated_rx.recv())
.await
.expect("timeout: push-update was not forwarded")
.unwrap();
assert!(
content_id.is_some(),
"with anydata_strict disabled, an unknown node inside the anydata payload must \
not cause validation to fail"
);
assert!(!sub_info.is_empty());

handle.shutdown().await.unwrap();
caching_handle.shutdown().await.unwrap();
caching_join_handle.await.unwrap().unwrap();
}

// TODO(libyang): mandatory-node enforcement inside `anydata` is not yet
// implemented upstream; once fixed, flip assertions to
// `res.is_err()` + `logs_contain`.
Expand Down