Merge pull request #15254 from LabNConsulting/chopps/notifications

Add YANG notifications

.gitignore

@@ -117,3 +117,4 @@ refix
 /test-suite.log
 pceplib/test/*.log
 pceplib/test/*.trs
+/tests/topotests/lib/mgmt_pb2.py

configure.ac

@@ -701,6 +701,8 @@ AC_ARG_ENABLE([mgmtd],
   AS_HELP_STRING([--disable-mgmtd], [do not build mgmtd]))
 AC_ARG_ENABLE([mgmtd_local_validations],
   AS_HELP_STRING([--enable-mgmtd-local-validations], [dev: unimplemented local validation]))
+AC_ARG_ENABLE([mgmtd_test_be_client],
+  AS_HELP_STRING([--enable-mgmtd-test-be-client], [build test backend client]))
 AC_ARG_ENABLE([ripd],
   AS_HELP_STRING([--disable-ripd], [do not build ripd]))
 AC_ARG_ENABLE([ripngd],
@@ -1811,6 +1813,10 @@ AS_IF([test "$enable_mgmtd" != "no"], [
   ])
 ])
 
+AS_IF([test "$enable_mgmtd_test_be_client" = "yes"], [
+  AC_DEFINE([HAVE_MGMTD_TESTC], [1], [mgmtd_testc])
+])
+
 AS_IF([test "$enable_ripd" != "no"], [
   AC_DEFINE([HAVE_RIPD], [1], [ripd])
 ])
@@ -2772,6 +2778,7 @@ AM_CONDITIONAL([VTYSH], [test "$VTYSH" = "vtysh"])
 AM_CONDITIONAL([ZEBRA], [test "$enable_zebra" != "no"])
 AM_CONDITIONAL([BGPD], [test "$enable_bgpd" != "no"])
 AM_CONDITIONAL([MGMTD], [test "$enable_mgmtd" != "no"])
+AM_CONDITIONAL([MGMTD_TESTC], [test "$enable_mgmtd_test_be_client" = "yes"])
 AM_CONDITIONAL([RIPD], [test "$enable_ripd" != "no"])
 AM_CONDITIONAL([OSPFD], [test "$enable_ospfd" != "no"])
 AM_CONDITIONAL([LDPD], [test "$enable_ldpd" != "no"])

doc/developer/northbound/retrofitting-configuration-commands.rst

@@ -982,7 +982,7 @@ future.
 For libfrr commands, it’s not possible to centralize all commands in a
 single file because the *extract.pl* script from *vtysh* treats commands
 differently depending on the file in which they are defined (e.g. DEFUNs
-from *lib/routemap.c* are installed using the ``VTYSH_RMAP`` constant,
+from *lib/routemap.c* are installed using the ``VTYSH_RMAP_SHOW`` constant,
 which identifies the daemons that support route-maps). In this case, the
 CLI commands should be rewritten but maintained in the same file.
 

doc/developer/topotests.rst

@@ -33,6 +33,7 @@ Installing Topotest Requirements
        tshark \
        valgrind
    python3 -m pip install wheel
+   python3 -m pip install protobuf
    python3 -m pip install 'pytest>=6.2.4'
    python3 -m pip install 'pytest-xdist>=2.3.0'
    python3 -m pip install 'scapy>=2.4.5'

lib/mgmt.proto

@@ -76,8 +76,9 @@ message YangGetDataReq {
 //
 message BeSubscribeReq {
   required string client_name = 1;
-  required bool subscribe_xpaths = 2;
+  repeated string config_xpaths = 2;
-  repeated string xpath_reg = 3;
+  repeated string oper_xpaths = 3;
+  repeated string notif_xpaths = 4;
 }
 
 message BeSubscribeReply {

lib/mgmt_be_client.c

@@ -311,6 +311,90 @@ static int be_client_send_error(struct mgmt_be_client *client, uint64_t txn_id,
 	return ret;
 }
 
+void mgmt_be_send_notification(struct lyd_node *tree)
+{
+	struct mgmt_be_client *client = __be_client;
+	struct mgmt_msg_notify_data *msg = NULL;
+	LYD_FORMAT format = LYD_JSON;
+	uint8_t **darrp;
+	LY_ERR err;
+
+	assert(tree);
+
+	MGMTD_BE_CLIENT_DBG("%s: sending YANG notification: %s", __func__,
+			    tree->schema->name);
+	/*
+	 * Allocate a message and append the data to it using `format`
+	 */
+	msg = mgmt_msg_native_alloc_msg(struct mgmt_msg_notify_data, 0,
+					MTYPE_MSG_NATIVE_NOTIFY);
+	msg->code = MGMT_MSG_CODE_NOTIFY;
+	msg->result_type = format;
+
+	darrp = mgmt_msg_native_get_darrp(msg);
+	err = yang_print_tree_append(darrp, tree, format,
+				     (LYD_PRINT_SHRINK | LYD_PRINT_WD_EXPLICIT |
+				      LYD_PRINT_WITHSIBLINGS));
+	if (err) {
+		flog_err(EC_LIB_LIBYANG,
+			 "%s: error creating notification data: %s", __func__,
+			 ly_strerrcode(err));
+		goto done;
+	}
+
+	(void)be_client_send_native_msg(client, msg,
+					mgmt_msg_native_get_msg_len(msg), false);
+done:
+	mgmt_msg_native_free_msg(msg);
+	lyd_free_all(tree);
+}
+
+/*
+ * Convert old style NB notification data into new MGMTD YANG tree and send.
+ */
+static int mgmt_be_notification_send(void *arg, const char *xpath,
+				     struct list *args)
+{
+	struct lyd_node *root = NULL;
+	struct lyd_node *dnode;
+	struct yang_data *data;
+	struct listnode *ln;
+	LY_ERR err;
+
+	MGMTD_BE_CLIENT_DBG("%s: sending notification: %s", __func__, xpath);
+
+	/*
+	 * Convert yang data args list to a libyang data tree
+	 */
+	for (ALL_LIST_ELEMENTS_RO(args, ln, data)) {
+		err = lyd_new_path(root, ly_native_ctx, data->xpath,
+				   data->value, LYD_NEW_PATH_UPDATE, &dnode);
+		if (err != LY_SUCCESS) {
+lyerr:
+			flog_err(EC_LIB_LIBYANG,
+				 "%s: error creating notification data: %s",
+				 __func__, ly_strerrcode(err));
+			if (root)
+				lyd_free_all(root);
+			return 1;
+		}
+		if (!root) {
+			root = dnode;
+			while (root->parent)
+				root = lyd_parent(root);
+		}
+	}
+
+	if (!root) {
+		err = lyd_new_path(NULL, ly_native_ctx, xpath, "", 0, &root);
+		if (err)
+			goto lyerr;
+	}
+
+	mgmt_be_send_notification(root);
+	return 0;
+}
+
 static int mgmt_be_send_txn_reply(struct mgmt_be_client *client_ctx,
 				  uint64_t txn_id, bool create)
 {
@@ -738,6 +822,12 @@ static int mgmt_be_client_handle_msg(struct mgmt_be_client *client_ctx,
 	case MGMTD__BE_MESSAGE__MESSAGE_SUBSCR_REPLY:
 		MGMTD_BE_CLIENT_DBG("Got SUBSCR_REPLY success %u",
 				    be_msg->subscr_reply->success);
+
+		if (client_ctx->cbs.subscr_done)
+			(*client_ctx->cbs.subscr_done)(client_ctx,
+						       client_ctx->user_data,
+						       be_msg->subscr_reply
+							       ->success);
 		break;
 	case MGMTD__BE_MESSAGE__MESSAGE_TXN_REQ:
 		MGMTD_BE_CLIENT_DBG("Got TXN_REQ %s txn-id: %" PRIu64,
@@ -824,7 +914,7 @@ static enum nb_error be_client_send_tree_data_batch(const struct lyd_node *tree,
 
 	darrp = mgmt_msg_native_get_darrp(tree_msg);
 	err = yang_print_tree_append(darrp, tree, args->result_type,
-				     (LYD_PRINT_WD_EXPLICIT |
+				     (LYD_PRINT_SHRINK | LYD_PRINT_WD_EXPLICIT |
 				      LYD_PRINT_WITHSIBLINGS));
 	if (err) {
 		ret = NB_ERR;
@@ -873,6 +963,31 @@ static void be_client_handle_get_tree(struct mgmt_be_client *client,
 		   be_client_send_tree_data_batch, args);
 }
 
+/*
+ * Process the notification.
+ */
+static void be_client_handle_notify(struct mgmt_be_client *client, void *msgbuf,
+				    size_t msg_len)
+{
+	struct mgmt_msg_notify_data *notif_msg = msgbuf;
+	struct mgmt_be_client_notification_cb *cb;
+	const char *notif;
+	uint i;
+
+	MGMTD_BE_CLIENT_DBG("Received notification for client %s", client->name);
+
+	/* "{\"modname:notification-name\": ...}" */
+	notif = (const char *)notif_msg->result + 2;
+
+	for (i = 0; i < client->cbs.nnotify_cbs; i++) {
+		cb = &client->cbs.notify_cbs[i];
+		if (strncmp(cb->xpath, notif, strlen(cb->xpath)))
+			continue;
+		cb->callback(client, client->user_data, cb,
+			     (const char *)notif_msg->result);
+	}
+}
+
 /*
  * Handle a native encoded message
  *
@@ -888,12 +1003,16 @@ static void be_client_handle_native_msg(struct mgmt_be_client *client,
 	case MGMT_MSG_CODE_GET_TREE:
 		be_client_handle_get_tree(client, txn_id, msg, msg_len);
 		break;
+	case MGMT_MSG_CODE_NOTIFY:
+		be_client_handle_notify(client, msg, msg_len);
+		break;
 	default:
 		MGMTD_BE_CLIENT_ERR("unknown native message txn-id %" PRIu64
 				    " req-id %" PRIu64 " code %u to client %s",
 				    txn_id, msg->req_id, msg->code,
 				    client->name);
-		be_client_send_error(client, msg->refer_id, msg->req_id, false, -1,
+		be_client_send_error(client, msg->refer_id, msg->req_id, false,
+				     -1,
 				     "BE cilent %s recv msg unknown txn-id %" PRIu64,
 				     client->name, txn_id);
 		break;
@@ -927,38 +1046,51 @@ static void mgmt_be_client_process_msg(uint8_t version, uint8_t *data,
 				    len);
 		return;
 	}
-	MGMTD_BE_CLIENT_DBG(
+	MGMTD_BE_CLIENT_DBG("Decoded %zu bytes of message(msg: %u/%u) from server",
-		"Decoded %zu bytes of message(msg: %u/%u) from server", len,
+			    len, be_msg->message_case, be_msg->message_case);
-		be_msg->message_case, be_msg->message_case);
 	(void)mgmt_be_client_handle_msg(client_ctx, be_msg);
 	mgmtd__be_message__free_unpacked(be_msg, NULL);
 }
 
 int mgmt_be_send_subscr_req(struct mgmt_be_client *client_ctx,
-			    bool subscr_xpaths, int num_xpaths,
+			    int n_config_xpaths, char **config_xpaths,
-			    char **reg_xpaths)
+			    int n_oper_xpaths, char **oper_xpaths)
 {
 	Mgmtd__BeMessage be_msg;
 	Mgmtd__BeSubscribeReq subscr_req;
+	const char **notif_xpaths = NULL;
+	int ret;
 
 	mgmtd__be_subscribe_req__init(&subscr_req);
 	subscr_req.client_name = client_ctx->name;
-	subscr_req.n_xpath_reg = num_xpaths;
+	subscr_req.n_config_xpaths = n_config_xpaths;
-	if (num_xpaths)
+	subscr_req.config_xpaths = config_xpaths;
-		subscr_req.xpath_reg = reg_xpaths;
+	subscr_req.n_oper_xpaths = n_oper_xpaths;
-	else
+	subscr_req.oper_xpaths = oper_xpaths;
-		subscr_req.xpath_reg = NULL;
+
-	subscr_req.subscribe_xpaths = subscr_xpaths;
+	/* See if we should register for notifications */
+	subscr_req.n_notif_xpaths = client_ctx->cbs.nnotify_cbs;
+	if (client_ctx->cbs.nnotify_cbs) {
+		struct mgmt_be_client_notification_cb *cb, *ecb;
+
+		cb = client_ctx->cbs.notify_cbs;
+		ecb = cb + client_ctx->cbs.nnotify_cbs;
+		for (; cb < ecb; cb++)
+			*darr_append(notif_xpaths) = cb->xpath;
+	}
+	subscr_req.notif_xpaths = (char **)notif_xpaths;
 
 	mgmtd__be_message__init(&be_msg);
 	be_msg.message_case = MGMTD__BE_MESSAGE__MESSAGE_SUBSCR_REQ;
 	be_msg.subscr_req = &subscr_req;
 
-	MGMTD_BE_CLIENT_DBG("Sending SUBSCR_REQ name: %s subscr_xpaths: %u num_xpaths: %zu",
+	MGMTD_BE_CLIENT_DBG("Sending SUBSCR_REQ name: %s xpaths: config %zu oper: %zu notif: %zu",
-			    subscr_req.client_name, subscr_req.subscribe_xpaths,
+			    subscr_req.client_name, subscr_req.n_config_xpaths,
-			    subscr_req.n_xpath_reg);
+			    subscr_req.n_oper_xpaths, subscr_req.n_notif_xpaths);
 
-	return mgmt_be_client_send_msg(client_ctx, &be_msg);
+	ret = mgmt_be_client_send_msg(client_ctx, &be_msg);
+	darr_free(notif_xpaths);
+	return ret;
 }
 
 static int _notify_conenct_disconnect(struct msg_client *msg_client,
@@ -970,15 +1102,16 @@ static int _notify_conenct_disconnect(struct msg_client *msg_client,
 
 	if (connected) {
 		assert(msg_client->conn.fd != -1);
-		ret = mgmt_be_send_subscr_req(client, false, 0, NULL);
+		ret = mgmt_be_send_subscr_req(client, 0, NULL, 0, NULL);
 		if (ret)
 			return ret;
 	}
 
 	/* Notify BE client through registered callback (if any) */
 	if (client->cbs.client_connect_notify)
-		(void)(*client->cbs.client_connect_notify)(
+		(void)(*client->cbs.client_connect_notify)(client,
-			client, client->user_data, connected);
+							   client->user_data,
+							   connected);
 
 	/* Cleanup any in-progress TXN on disconnect */
 	if (!connected)
@@ -1016,9 +1149,8 @@ static void mgmt_debug_client_be_set(uint32_t flags, bool set)
 
 DEFPY(debug_mgmt_client_be, debug_mgmt_client_be_cmd,
       "[no] debug mgmt client backend",
-      NO_STR DEBUG_STR MGMTD_STR
+      NO_STR DEBUG_STR MGMTD_STR "client\n"
-      "client\n"
+				 "backend\n")
-      "backend\n")
 {
 	mgmt_debug_client_be_set(DEBUG_NODE2MODE(vty->node), !no);
 
@@ -1083,6 +1215,10 @@ struct mgmt_be_client *mgmt_be_client_create(const char *client_name,
 			MGMTD_BE_MAX_NUM_MSG_WRITE, MGMTD_BE_MAX_MSG_LEN, false,
 			"BE-client", MGMTD_DBG_BE_CLIENT_CHECK());
 
+	/* Hook to receive notifications */
+	hook_register_arg(nb_notification_send, mgmt_be_notification_send,
+			  client);
+
 	MGMTD_BE_CLIENT_DBG("Initialized client '%s'", client_name);
 
 	return client;

lib/mgmt_be_client.h

@@ -60,14 +60,29 @@ struct mgmt_be_client_txn_ctx {
  * Callbacks:
  *	client_connect_notify: called when connection is made/lost to mgmtd.
  *	txn_notify: called when a txn has been created
+ *	notify_cbs: callbacks for notifications.
+ *	nnotify_cbs: number of notification callbacks.
+ *
  */
 struct mgmt_be_client_cbs {
 	void (*client_connect_notify)(struct mgmt_be_client *client,
 				      uintptr_t usr_data, bool connected);
-
+	void (*subscr_done)(struct mgmt_be_client *client, uintptr_t usr_data,
+			    bool success);
 	void (*txn_notify)(struct mgmt_be_client *client, uintptr_t usr_data,
 			   struct mgmt_be_client_txn_ctx *txn_ctx,
 			   bool destroyed);
+
+	struct mgmt_be_client_notification_cb *notify_cbs;
+	uint nnotify_cbs;
+};
+
+struct mgmt_be_client_notification_cb {
+	const char *xpath; /* the notification */
+	uint8_t format;	   /* currently only LYD_JSON supported */
+	void (*callback)(struct mgmt_be_client *client, uintptr_t usr_data,
+			 struct mgmt_be_client_notification_cb *this,
+			 const char *notif_data);
 };
 
 /***************************************************************
@@ -124,7 +139,7 @@ extern void mgmt_debug_be_client_show_debug(struct vty *vty);
  *    The client object.
  *
  * reg_yang_xpaths
- *    Yang xpath(s) that needs to be [un]-subscribed from/to
+ *    Yang xpath(s) that needs to be subscribed to
  *
  * num_xpaths
  *    Number of xpaths
@@ -132,9 +147,18 @@ extern void mgmt_debug_be_client_show_debug(struct vty *vty);
  * Returns:
  *    MGMTD_SUCCESS on success, MGMTD_* otherwise.
  */
-extern int mgmt_be_send_subscr_req(struct mgmt_be_client *client,
+extern int mgmt_be_send_subscr_req(struct mgmt_be_client *client_ctx,
-				   bool subscr_xpaths, int num_xpaths,
+				   int n_config_xpaths, char **config_xpaths,
-				   char **reg_xpaths);
+				   int n_oper_xpaths, char **oper_xpaths);
+
+/**
+ * mgmt_be_notification_send() - send a YANG notification to FE clients.
+ * @tree: libyang tree for the notification. The tree will be freed by
+ *        this function.
+ *
+ */
+extern void mgmt_be_send_notification(struct lyd_node *tree);
+
 
 /*
  * Destroy backend client and cleanup everything.

lib/mgmt_fe_client.c

@@ -507,19 +507,24 @@ static void fe_client_handle_native_msg(struct mgmt_fe_client *client,
 					struct mgmt_msg_header *msg,
 					size_t msg_len)
 {
-	struct mgmt_fe_client_session *session;
+	struct mgmt_fe_client_session *session = NULL;
+	struct mgmt_msg_notify_data *notify_msg;
 	struct mgmt_msg_tree_data *tree_msg;
 	struct mgmt_msg_error *err_msg;
+	char *notify_data = NULL;
 
-	MGMTD_FE_CLIENT_DBG("Got GET_TREE reply for session-id %" PRIu64,
+	MGMTD_FE_CLIENT_DBG("Got native message for session-id %" PRIu64,
 			    msg->refer_id);
 
-	session = mgmt_fe_find_session_by_session_id(client, msg->refer_id);
+	if (msg->code != MGMT_MSG_CODE_NOTIFY) {
-
+		session = mgmt_fe_find_session_by_session_id(client,
-	if (!session || !session->client) {
+							     msg->refer_id);
-		MGMTD_FE_CLIENT_ERR("No session for received native msg session-id %" PRIu64,
+		if (!session || !session->client) {
-				    msg->refer_id);
+			MGMTD_FE_CLIENT_ERR(
-		return;
+				"No session for received native msg session-id %" PRIu64,
+				msg->refer_id);
+			return;
+		}
 	}
 
 	switch (msg->code) {
@@ -559,6 +564,44 @@ static void fe_client_handle_native_msg(struct mgmt_fe_client *client,
 						     msg_len - sizeof(*tree_msg),
 						     tree_msg->partial_error);
 		break;
+	case MGMT_MSG_CODE_NOTIFY:
+		notify_msg = (typeof(notify_msg))msg;
+		if (msg_len < sizeof(*notify_msg)) {
+			MGMTD_FE_CLIENT_ERR("Corrupt notify-data msg recv");
+			return;
+		}
+
+		if (notify_msg->result_type != LYD_LYB &&
+		    !MGMT_MSG_VALIDATE_NUL_TERM(notify_msg, msg_len)) {
+			MGMTD_FE_CLIENT_ERR("Corrupt error msg recv");
+			return;
+		}
+		if (notify_msg->result_type == LYD_JSON)
+			notify_data = (char *)notify_msg->result;
+		else
+			notify_data =
+				yang_convert_lyd_format(notify_msg->result,
+							msg_len,
+							notify_msg->result_type,
+							LYD_JSON, true);
+		if (!notify_data) {
+			MGMTD_FE_CLIENT_ERR("Can't convert format %d to JSON",
+					    notify_msg->result_type);
+			return;
+		}
+		FOREACH_SESSION_IN_LIST (client, session) {
+			if (!session->client->cbs.async_notification)
+				continue;
+
+			session->client->cbs
+				.async_notification(client, client->user_data,
+						    session->client_id,
+						    session->user_ctx,
+						    notify_data);
+		}
+		if (notify_msg->result_type != LYD_JSON)
+			darr_free(notify_data);
+		break;
 	default:
 		MGMTD_FE_CLIENT_ERR("unknown native message session-id %" PRIu64
 				    " req-id %" PRIu64 " code %u",

lib/mgmt_fe_client.h

@@ -114,6 +114,11 @@ struct mgmt_fe_client_cbs {
 			       LYD_FORMAT result_type, void *result, size_t len,
 			       int partial_error);
 
+	/* Called with asynchronous notifications from backends */
+	int (*async_notification)(struct mgmt_fe_client *client,
+				  uintptr_t user_data, uint64_t client_id,
+				  uintptr_t session_ctx, const char *result);
+
 	/* Called when new native error is returned */
 	int (*error_notify)(struct mgmt_fe_client *client, uintptr_t user_data,
 			    uint64_t client_id, uint64_t session_id,

lib/mgmt_msg_native.c

@@ -14,6 +14,7 @@ DEFINE_MTYPE(MSG_NATIVE, MSG_NATIVE_ERROR, "native error msg");
 DEFINE_MTYPE(MSG_NATIVE, MSG_NATIVE_GET_TREE, "native get tree msg");
 DEFINE_MTYPE(MSG_NATIVE, MSG_NATIVE_TREE_DATA, "native tree data msg");
 DEFINE_MTYPE(MSG_NATIVE, MSG_NATIVE_GET_DATA, "native get data msg");
+DEFINE_MTYPE(MSG_NATIVE, MSG_NATIVE_NOTIFY, "native get data msg");
 
 int vmgmt_msg_native_send_error(struct msg_conn *conn, uint64_t sess_or_txn_id,
 				uint64_t req_id, bool short_circuit_ok,

lib/mgmt_msg_native.h

@@ -143,6 +143,7 @@ DECLARE_MTYPE(MSG_NATIVE_ERROR);
 DECLARE_MTYPE(MSG_NATIVE_GET_TREE);
 DECLARE_MTYPE(MSG_NATIVE_TREE_DATA);
 DECLARE_MTYPE(MSG_NATIVE_GET_DATA);
+DECLARE_MTYPE(MSG_NATIVE_NOTIFY);
 
 /*
  * Native message codes
@@ -151,6 +152,7 @@ DECLARE_MTYPE(MSG_NATIVE_GET_DATA);
 #define MGMT_MSG_CODE_GET_TREE	1
 #define MGMT_MSG_CODE_TREE_DATA 2
 #define MGMT_MSG_CODE_GET_DATA	3
+#define MGMT_MSG_CODE_NOTIFY	4
 
 /**
  * struct mgmt_msg_header - Header common to all native messages.
@@ -257,8 +259,29 @@ _Static_assert(sizeof(struct mgmt_msg_get_data) ==
 		       offsetof(struct mgmt_msg_get_data, xpath),
 	       "Size mismatch");
 
+/**
+ * struct mgmt_msg_notify_data - Message carrying notification data.
+ *
+ * @result_type: ``LYD_FORMAT`` for format of the @result value.
+ * @result: The tree data in @result_type format.
+ *
+ */
+struct mgmt_msg_notify_data {
+	struct mgmt_msg_header;
+	uint8_t result_type;
+	uint8_t resv2[7];
+
+	alignas(8) uint8_t result[];
+};
+_Static_assert(sizeof(struct mgmt_msg_notify_data) ==
+		       offsetof(struct mgmt_msg_notify_data, result),
+	       "Size mismatch");
+
+/*
+ * Validate that the message ends in a NUL terminating byte
+ */
 #define MGMT_MSG_VALIDATE_NUL_TERM(msgp, len)                                  \
-	((len) >= sizeof(*msg) + 1 && ((char *)msgp)[(len)-1] == 0)
+	((len) >= sizeof(*msgp) + 1 && ((char *)msgp)[(len)-1] == 0)
 
 
 /**

lib/northbound.c

@@ -2068,6 +2068,23 @@ int nb_notification_send(const char *xpath, struct list *arguments)
 	return ret;
 }
 
+DEFINE_HOOK(nb_notification_tree_send, (struct lyd_node *tree), (tree));
+
+int nb_notification_tree_send(struct lyd_node *tree)
+{
+	int ret;
+
+	assert(tree);
+
+	DEBUGD(&nb_dbg_notif, "northbound tree notification: %s",
+	       tree->schema->name);
+
+	ret = hook_call(nb_notification_tree_send, tree);
+	lyd_free_all(tree);
+
+	return ret;
+}
+
 /* Running configuration user pointers management. */
 struct nb_config_entry {
 	char xpath[XPATH_MAXLEN];

lib/northbound.h

@@ -1441,6 +1441,10 @@ extern bool nb_cb_operation_is_valid(enum nb_cb_operation operation,
 				     const struct lysc_node *snode);
 
 /*
+ * DEPRECATED: This call and infra should no longer be used. Instead,
+ * the mgmtd supported tree based call `nb_notification_tree_send` should be
+ * used instead
+ *
  * Send a YANG notification. This is a no-op unless the 'nb_notification_send'
  * hook was registered by a northbound plugin.
  *
@@ -1456,6 +1460,19 @@ extern bool nb_cb_operation_is_valid(enum nb_cb_operation operation,
  */
 extern int nb_notification_send(const char *xpath, struct list *arguments);
 
+/*
+ * Send a YANG notification from a backend . This is a no-op unless th
+ * 'nb_notification_tree_send' hook was registered by a northbound plugin.
+ *
+ * tree
+ *    The libyang tree for the notification. The tree will be freed by
+ *    this call.
+ *
+ * Returns:
+ *    NB_OK on success, NB_ERR otherwise.
+ */
+extern int nb_notification_tree_send(struct lyd_node *tree);
+
 /*
  * Associate a user pointer to a configuration node.
  *

lib/yang.c

@@ -744,6 +744,34 @@ uint8_t *yang_print_tree(const struct lyd_node *root, LYD_FORMAT format,
 	return darr;
 }
 
+char *yang_convert_lyd_format(const uint8_t *data, size_t data_len,
+				     LYD_FORMAT in_format,
+				     LYD_FORMAT out_format, bool shrink)
+{
+	struct lyd_node *tree = NULL;
+	uint8_t *result = NULL;
+	uint32_t options = LYD_PRINT_WD_EXPLICIT | LYD_PRINT_WITHSIBLINGS;
+
+	assert(out_format != LYD_LYB);
+
+	if (!MGMT_MSG_VALIDATE_NUL_TERM(data, data_len))
+		return NULL;
+
+	if (in_format == out_format)
+		return darr_strdup((const char *)data);
+
+	if (shrink)
+		options |= LYD_PRINT_SHRINK;
+
+	/* Take a guess at the initial capacity based on input data size */
+	darr_ensure_cap(result, data_len);
+	if (yang_print_tree_append(&result, tree, out_format, options)) {
+		darr_free(result);
+		return NULL;
+	}
+	return (char *)result;
+}
+
 const char *yang_print_errors(struct ly_ctx *ly_ctx, char *buf, size_t buf_len)
 {
 	struct ly_err_item *ei;

lib/yang.h

@@ -622,6 +622,22 @@ extern void yang_debugging_set(bool enable);
 extern uint8_t *yang_print_tree(const struct lyd_node *root, LYD_FORMAT format,
 				uint32_t options);
 
+
+/**
+ * yang_convert_lyd_format() - convert one libyang format to darr string.
+ * @data: data to convert.
+ * @data_len: length of the data.
+ * @in_format: format of the data.
+ * @out_format: format to return.
+ * @shrink: true to avoid pretty printing.
+ *
+ * Return:
+ *	A darr based string or NULL for error.
+ */
+extern char *yang_convert_lyd_format(const uint8_t *data, size_t msg_len,
+				     LYD_FORMAT in_format,
+				     LYD_FORMAT out_format, bool shrink);
+
 /*
  * "Print" the yang tree in `root` into an existing dynamic sized array.
  *

mgmtd/mgmt_be_adapter.c

@@ -35,6 +35,7 @@
 /* ---------- */
 
 const char *mgmt_be_client_names[MGMTD_BE_CLIENT_ID_MAX + 1] = {
+	[MGMTD_BE_CLIENT_ID_TESTC] = "mgmtd-testc", /* always first */
 	[MGMTD_BE_CLIENT_ID_ZEBRA] = "zebra",
 #ifdef HAVE_RIPD
 	[MGMTD_BE_CLIENT_ID_RIPD] = "ripd",
@@ -155,6 +156,7 @@ static const char *const *be_client_oper_xpaths[MGMTD_BE_CLIENT_ID_MAX] = {
 
 static struct mgmt_be_xpath_map *be_cfg_xpath_map;
 static struct mgmt_be_xpath_map *be_oper_xpath_map;
+static struct mgmt_be_xpath_map *be_notif_xpath_map;
 
 static struct event_loop *mgmt_loop;
 static struct msg_server mgmt_be_server = {.fd = -1};
@@ -219,11 +221,16 @@ mgmt_be_find_adapter_by_name(const char *name)
 }
 
 static void mgmt_register_client_xpath(enum mgmt_be_client_id id,
-				       const char *xpath, bool config)
+				       const char *xpath, bool config, bool oper)
 {
 	struct mgmt_be_xpath_map **maps, *map;
 
-	maps = config ? &be_cfg_xpath_map : &be_oper_xpath_map;
+	if (config)
+		maps = &be_cfg_xpath_map;
+	else if (oper)
+		maps = &be_oper_xpath_map;
+	else
+		maps = &be_notif_xpath_map;
 
 	darr_foreach_p (*maps, map) {
 		if (!strcmp(xpath, map->xpath_prefix)) {
@@ -251,13 +258,13 @@ static void mgmt_be_xpath_map_init(void)
 		/* Initialize the common config init map */
 		for (init = be_client_config_xpaths[id]; init && *init; init++) {
 			MGMTD_BE_ADAPTER_DBG(" - CFG XPATH: '%s'", *init);
-			mgmt_register_client_xpath(id, *init, true);
+			mgmt_register_client_xpath(id, *init, true, false);
 		}
 
 		/* Initialize the common oper init map */
 		for (init = be_client_oper_xpaths[id]; init && *init; init++) {
 			MGMTD_BE_ADAPTER_DBG(" - OPER XPATH: '%s'", *init);
-			mgmt_register_client_xpath(id, *init, false);
+			mgmt_register_client_xpath(id, *init, false, true);
 		}
 	}
 
@@ -278,6 +285,10 @@ static void mgmt_be_xpath_map_cleanup(void)
 	darr_foreach_p (be_oper_xpath_map, map)
 		XFREE(MTYPE_MGMTD_XPATH, map->xpath_prefix);
 	darr_free(be_oper_xpath_map);
+
+	darr_foreach_p (be_notif_xpath_map, map)
+		XFREE(MTYPE_MGMTD_XPATH, map->xpath_prefix);
+	darr_free(be_notif_xpath_map);
 }
 
 
@@ -388,20 +399,20 @@ static int
 mgmt_be_adapter_handle_msg(struct mgmt_be_client_adapter *adapter,
 			      Mgmtd__BeMessage *be_msg)
 {
+	const char *xpath;
+	uint i, num;
+
 	/*
 	 * protobuf-c adds a max size enum with an internal, and changing by
 	 * version, name; cast to an int to avoid unhandled enum warnings
 	 */
 	switch ((int)be_msg->message_case) {
 	case MGMTD__BE_MESSAGE__MESSAGE_SUBSCR_REQ:
-		MGMTD_BE_ADAPTER_DBG(
+		MGMTD_BE_ADAPTER_DBG("Got SUBSCR_REQ from '%s' to register xpaths config: %zu oper: %zu notif: %zu",
-			"Got SUBSCR_REQ from '%s' to %sregister %zu xpaths",
+				     be_msg->subscr_req->client_name,
-			be_msg->subscr_req->client_name,
+				     be_msg->subscr_req->n_config_xpaths,
-			!be_msg->subscr_req->subscribe_xpaths &&
+				     be_msg->subscr_req->n_oper_xpaths,
-					be_msg->subscr_req->n_xpath_reg
+				     be_msg->subscr_req->n_notif_xpaths);
-				? "de"
-				: "",
-			be_msg->subscr_req->n_xpath_reg);
 
 		if (strlen(be_msg->subscr_req->client_name)) {
 			strlcpy(adapter->name, be_msg->subscr_req->client_name,
@@ -413,7 +424,6 @@ mgmt_be_adapter_handle_msg(struct mgmt_be_client_adapter *adapter,
 					adapter->name);
 				/* this will/should delete old */
 				msg_conn_disconnect(adapter->conn, false);
-				zlog_err("XXX different from original code");
 				break;
 			}
 			mgmt_be_adapters_by_id[adapter->id] = adapter;
@@ -423,11 +433,28 @@ mgmt_be_adapter_handle_msg(struct mgmt_be_client_adapter *adapter,
 			mgmt_be_adapter_sched_init_event(adapter);
 		}
 
-		if (be_msg->subscr_req->n_xpath_reg)
+		num = be_msg->subscr_req->n_config_xpaths;
-			/* we aren't handling dynamic xpaths yet */
+		for (i = 0; i < num; i++) {
-			mgmt_be_send_subscr_reply(adapter, false);
+			xpath = be_msg->subscr_req->config_xpaths[i];
-		else
+			mgmt_register_client_xpath(adapter->id, xpath, true,
-			mgmt_be_send_subscr_reply(adapter, true);
+						   false);
+		}
+
+		num = be_msg->subscr_req->n_oper_xpaths;
+		for (i = 0; i < num; i++) {
+			xpath = be_msg->subscr_req->oper_xpaths[i];
+			mgmt_register_client_xpath(adapter->id, xpath, false,
+						   true);
+		}
+
+		num = be_msg->subscr_req->n_notif_xpaths;
+		for (i = 0; i < num; i++) {
+			xpath = be_msg->subscr_req->notif_xpaths[i];
+			mgmt_register_client_xpath(adapter->id, xpath, false,
+						   false);
+		}
+
+		mgmt_be_send_subscr_reply(adapter, true);
 		break;
 	case MGMTD__BE_MESSAGE__MESSAGE_TXN_REPLY:
 		MGMTD_BE_ADAPTER_DBG(
@@ -575,6 +602,34 @@ int mgmt_be_send_native(enum mgmt_be_client_id id, void *msg)
 	return mgmt_msg_native_send_msg(adapter->conn, msg, false);
 }
 
+static void mgmt_be_adapter_send_notify(struct mgmt_msg_notify_data *msg,
+					size_t msglen)
+{
+	struct mgmt_be_client_adapter *adapter;
+	struct mgmt_be_xpath_map *map;
+	const char *notif;
+	uint id;
+
+	if (!darr_len(be_notif_xpath_map))
+		return;
+
+	/* "{\"modname:notification-name\": ...}" */
+	notif = (const char *)msg->result + 2;
+
+	darr_foreach_p (be_notif_xpath_map, map) {
+		if (strncmp(map->xpath_prefix, notif, strlen(map->xpath_prefix)))
+			continue;
+
+		FOREACH_BE_CLIENT_BITS (id, map->clients) {
+			adapter = mgmt_be_get_adapter_by_id(id);
+			if (!adapter)
+				continue;
+			msg_conn_send_msg(adapter->conn, MGMT_MSG_VERSION_NATIVE,
+					  msg, msglen, NULL, false);
+		}
+	}
+}
+
 /*
  * Handle a native encoded message
  */
@@ -582,6 +637,7 @@ static void be_adapter_handle_native_msg(struct mgmt_be_client_adapter *adapter,
 					 struct mgmt_msg_header *msg,
 					 size_t msg_len)
 {
+	struct mgmt_msg_notify_data *notify_msg;
 	struct mgmt_msg_tree_data *tree_msg;
 	struct mgmt_msg_error *error_msg;
 
@@ -607,6 +663,12 @@ static void be_adapter_handle_native_msg(struct mgmt_be_client_adapter *adapter,
 		/* Forward the reply to the txn module */
 		mgmt_txn_notify_tree_data_reply(adapter, tree_msg, msg_len);
 		break;
+	case MGMT_MSG_CODE_NOTIFY:
+		notify_msg = (typeof(notify_msg))msg;
+		MGMTD_BE_ADAPTER_DBG("Got NOTIFY from '%s'", adapter->name);
+		mgmt_be_adapter_send_notify(notify_msg, msg_len);
+		mgmt_fe_adapter_send_notify(notify_msg, msg_len);
+		break;
 	default:
 		MGMTD_BE_ADAPTER_ERR("unknown native message txn-id %" PRIu64
 				     " req-id %" PRIu64

mgmtd/mgmt_be_adapter.h

@@ -27,6 +27,8 @@
  * #ifdef HAVE_COMPONENT
  */
 enum mgmt_be_client_id {
+	MGMTD_BE_CLIENT_ID_TESTC, /* always first */
+	MGMTD_BE_CLIENT_ID_ZEBRA,
 #ifdef HAVE_RIPD
 	MGMTD_BE_CLIENT_ID_RIPD,
 #endif
@@ -36,7 +38,6 @@ enum mgmt_be_client_id {
 #ifdef HAVE_STATICD
 	MGMTD_BE_CLIENT_ID_STATICD,
 #endif
-	MGMTD_BE_CLIENT_ID_ZEBRA,
 	MGMTD_BE_CLIENT_ID_MAX
 };
 #define MGMTD_BE_CLIENT_ID_MIN	0
@@ -244,6 +245,13 @@ extern int mgmt_be_send_native(enum mgmt_be_client_id id, void *msg);
  */
 extern uint64_t mgmt_be_interested_clients(const char *xpath, bool config);
 
+/**
+ * mgmt_fe_adapter_send_notify() - notify FE clients of a notification.
+ * @msg: the notify message from the backend client.
+ * @msglen: the length of the notify message.
+ */
+extern void mgmt_fe_adapter_send_notify(struct mgmt_msg_notify_data *msg,
+					size_t msglen);
 /*
  * Dump backend client information for a given xpath to vty.
  */

mgmtd/mgmt_fe_adapter.c

@@ -1087,11 +1087,11 @@ static int fe_adapter_send_tree_data(struct mgmt_fe_session_ctx *session,
 {
 	struct mgmt_msg_tree_data *msg;
 	struct lyd_node *empty = NULL;
-	uint8_t *buf = NULL;
+	uint8_t **darrp = NULL;
 	int ret = 0;
 
-	darr_append_n(buf, sizeof(*msg));
+	msg = mgmt_msg_native_alloc_msg(struct mgmt_msg_tree_data, 0,
-	msg = (typeof(msg))buf;
+					MTYPE_MSG_NATIVE_TREE_DATA);
 	msg->refer_id = session->session_id;
 	msg->req_id = req_id;
 	msg->code = MGMT_MSG_CODE_TREE_DATA;
@@ -1103,13 +1103,10 @@ static int fe_adapter_send_tree_data(struct mgmt_fe_session_ctx *session,
 		tree = empty;
 	}
 
-	ret = yang_print_tree_append(&buf, tree, result_type,
+	darrp = mgmt_msg_native_get_darrp(msg);
+	ret = yang_print_tree_append(darrp, tree, result_type,
 				     (LYD_PRINT_WD_EXPLICIT |
 				      LYD_PRINT_WITHSIBLINGS));
-	/* buf may have been reallocated and moved */
-	msg = (typeof(msg))buf;
-	(void)msg; /* suppress clang-SA unused warning on safety code */
-
 	if (ret != LY_SUCCESS) {
 		MGMTD_FE_ADAPTER_ERR("Error building get-tree result for client %s session-id %" PRIu64
 				     " req-id %" PRIu64
@@ -1121,15 +1118,17 @@ static int fe_adapter_send_tree_data(struct mgmt_fe_session_ctx *session,
 
 	MGMTD_FE_ADAPTER_DBG("Sending get-tree result from adapter %s to session-id %" PRIu64
 			     " req-id %" PRIu64 " scok %d result type %u len %u",
-			     session->adapter->name, session->session_id, req_id,
+			     session->adapter->name, session->session_id,
-			     short_circuit_ok, result_type, darr_len(buf));
+			     req_id, short_circuit_ok, result_type,
+			     mgmt_msg_native_get_msg_len(msg));
 
-	ret = fe_adapter_send_native_msg(session->adapter, buf, darr_len(buf),
+	ret = fe_adapter_send_native_msg(session->adapter, msg,
+					 mgmt_msg_native_get_msg_len(msg),
 					 short_circuit_ok);
 done:
 	if (empty)
 		yang_dnode_free(empty);
-	darr_free(buf);
+	mgmt_msg_native_free_msg(msg);
 
 	return ret;
 }
@@ -1286,6 +1285,23 @@ static void mgmt_fe_adapter_process_msg(uint8_t version, uint8_t *data,
 	mgmtd__fe_message__free_unpacked(fe_msg, NULL);
 }
 
+void mgmt_fe_adapter_send_notify(struct mgmt_msg_notify_data *msg, size_t msglen)
+{
+	struct mgmt_fe_client_adapter *adapter;
+	struct mgmt_fe_session_ctx *session;
+
+	assert(msg->refer_id == 0);
+
+	FOREACH_ADAPTER_IN_LIST (adapter) {
+		FOREACH_SESSION_IN_LIST (adapter, session) {
+			msg->refer_id = session->session_id;
+			(void)fe_adapter_send_native_msg(adapter, msg, msglen,
+							 false);
+		}
+	}
+	msg->refer_id = 0;
+}
+
 void mgmt_fe_adapter_lock(struct mgmt_fe_client_adapter *adapter)
 {
 	adapter->refcount++;

mgmtd/mgmt_testc.c

@@ -0,0 +1,155 @@
+// SPDX-License-Identifier: GPL-2.0-or-later
+/*
+ * January 29 2024, Christian Hopps <chopps@labn.net>
+ *
+ * Copyright (c) 2024, LabN Consulting, L.L.C.
+ *
+ */
+
+#include <zebra.h>
+#include <lib/version.h>
+#include "libfrr.h"
+#include "mgmt_be_client.h"
+
+/* ---------------- */
+/* Local Prototypes */
+/* ---------------- */
+
+static void ripd_notification(struct mgmt_be_client *client, uintptr_t usr_data,
+			      struct mgmt_be_client_notification_cb *this,
+			      const char *notif_data);
+
+static void sigusr1(void);
+static void sigint(void);
+
+/* ----------- */
+/* Global Data */
+/* ----------- */
+
+/* privileges */
+static zebra_capabilities_t _caps_p[] = {};
+
+struct zebra_privs_t __privs = {
+#if defined(FRR_USER) && defined(FRR_GROUP)
+	.user = FRR_USER,
+	.group = FRR_GROUP,
+#endif
+#ifdef VTY_GROUP
+	.vty_group = VTY_GROUP,
+#endif
+	.caps_p = _caps_p,
+	.cap_num_p = array_size(_caps_p),
+	.cap_num_i = 0,
+};
+
+struct option longopts[] = {{0}};
+
+/* Master of threads. */
+struct event_loop *master;
+
+struct mgmt_be_client *mgmt_be_client;
+
+static struct frr_daemon_info mgmtd_testc_di;
+
+struct frr_signal_t __signals[] = {
+	{
+		.signal = SIGUSR1,
+		.handler = &sigusr1,
+	},
+	{
+		.signal = SIGINT,
+		.handler = &sigint,
+	},
+	{
+		.signal = SIGTERM,
+		.handler = &sigint,
+	},
+};
+
+#define MGMTD_TESTC_VTY_PORT 2624
+
+/* clang-format off */
+FRR_DAEMON_INFO(mgmtd_testc, MGMTD_TESTC,
+		.proghelp = "FRR Management Daemon Test Client.",
+
+		.signals = __signals,
+		.n_signals = array_size(__signals),
+
+		.privs = &__privs,
+
+		// .yang_modules = mgmt_yang_modules,
+		// .n_yang_modules = array_size(mgmt_yang_modules),
+
+		/* avoid libfrr trying to read our config file for us */
+		.flags = FRR_MANUAL_VTY_START,
+	);
+/* clang-format on */
+
+struct mgmt_be_client_notification_cb __notify_cbs[] = { {
+	.xpath = "frr-ripd",
+	.format = LYD_JSON,
+	.callback = ripd_notification,
+} };
+
+struct mgmt_be_client_cbs __client_cbs = {
+	.notify_cbs = __notify_cbs,
+	.nnotify_cbs = array_size(__notify_cbs),
+};
+
+
+/* --------- */
+/* Functions */
+/* --------- */
+
+
+static void sigusr1(void)
+{
+	zlog_rotate();
+}
+
+static void sigint(void)
+{
+	zlog_notice("Terminating on signal");
+	frr_fini();
+	exit(0);
+}
+
+static void ripd_notification(struct mgmt_be_client *client, uintptr_t usr_data,
+			      struct mgmt_be_client_notification_cb *this,
+			      const char *notif_data)
+{
+	zlog_notice("Received RIPd notification");
+}
+
+int main(int argc, char **argv)
+{
+	frr_preinit(&mgmtd_testc_di, argc, argv);
+	frr_opt_add("", longopts, "");
+
+	while (1) {
+		int opt;
+
+		opt = frr_getopt(argc, argv, NULL);
+
+		if (opt == EOF)
+			break;
+
+		switch (opt) {
+		case 0:
+			break;
+		default:
+			frr_help_exit(1);
+		}
+	}
+
+	master = frr_init();
+
+	mgmt_be_client = mgmt_be_client_create("mgmtd-testc", &__client_cbs, 0,
+					       master);
+
+	frr_config_fork();
+	frr_run(master);
+
+	/* Reached. */
+	return 0;
+}

mgmtd/subdir.am

@@ -50,6 +50,12 @@ noinst_HEADERS += \
 
 sbin_PROGRAMS += mgmtd/mgmtd
 
+if MGMTD_TESTC
+sbin_PROGRAMS += mgmtd/mgmtd_testc
+mgmtd_mgmtd_testc_SOURCES = mgmtd/mgmt_testc.c
+mgmtd_mgmtd_testc_LDADD = lib/libfrr.la
+endif
+
 mgmtd_mgmtd_SOURCES = \
 	mgmtd/mgmt_main.c \
 	# end

tests/topotests/lib/fe_client.py

@@ -0,0 +1,414 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 eval: (blacken-mode 1) -*-
+# SPDX-License-Identifier: GPL-2.0-or-later
+#
+# November 27 2023, Christian Hopps <chopps@labn.net>
+#
+# Copyright (c) 2023, LabN Consulting, L.L.C.
+#
+# noqa: E501
+#
+import argparse
+import json
+import logging
+import os
+import socket
+import struct
+import sys
+import time
+from pathlib import Path
+
+CWD = os.path.dirname(os.path.realpath(__file__))
+
+# This is painful but works if you have installed protobuf would be better if we
+# actually built and installed these but ... python packaging.
+try:
+    sys.path.append(os.path.dirname(CWD))
+    from munet.base import commander
+
+    commander.cmd_raises(f"protoc --python_out={CWD} -I {CWD}/../../../lib mgmt.proto")
+except Exception as error:
+    logging.error("can't create protobuf definition modules %s", error)
+    raise
+
+try:
+    sys.path[0:0] = "."
+    import mgmt_pb2
+except Exception as error:
+    logging.error("can't import proto definition modules %s", error)
+    raise
+
+CANDIDATE_DS = mgmt_pb2.DatastoreId.CANDIDATE_DS
+OPERATIONAL_DS = mgmt_pb2.DatastoreId.OPERATIONAL_DS
+RUNNING_DS = mgmt_pb2.DatastoreId.RUNNING_DS
+STARTUP_DS = mgmt_pb2.DatastoreId.STARTUP_DS
+
+# =====================
+# Native message values
+# =====================
+
+MGMT_MSG_MARKER_PROTOBUF = b"\000###"
+MGMT_MSG_MARKER_NATIVE = b"\001###"
+
+#
+# Native message formats
+#
+MSG_HDR_FMT = "=H2xIQQ"
+HDR_FIELD_CODE = 0
+HDR_FIELD_VSPLIT = 1
+HDR_FIELD_SESS_ID = 2
+HDR_FIELD_REQ_ID = 3
+
+MSG_ERROR_FMT = "=h6x"
+ERROR_FIELD_ERROR = 0
+
+# MSG_GET_TREE_FMT = "=B7x"
+# GET_TREE_FIELD_RESULT_TYPE = 0
+
+MSG_TREE_DATA_FMT = "=bBB5x"
+TREE_DATA_FIELD_PARTIAL_ERROR = 0
+TREE_DATA_FIELD_RESULT_TYPE = 1
+TREE_DATA_FIELD_MORE = 2
+
+MSG_GET_DATA_FMT = "=BB6x"
+GET_DATA_FIELD_RESULT_TYPE = 0
+GET_DATA_FIELD_FLAGS = 1
+GET_DATA_FLAG_STATE = 0x1
+GET_DATA_FLAG_CONFIG = 0x2
+GET_DATA_FLAG_EXACT = 0x4
+
+MSG_NOTIFY_FMT = "=B7x"
+NOTIFY_FIELD_RESULT_TYPE = 0
+
+#
+# Native message codes
+#
+MSG_CODE_ERROR = 0
+# MSG_CODE_GET_TREE = 1
+MSG_CODE_TREE_DATA = 2
+MSG_CODE_GET_DATA = 3
+MSG_CODE_NOTIFY = 4
+
+msg_native_formats = {
+    MSG_CODE_ERROR: MSG_ERROR_FMT,
+    # MSG_CODE_GET_TREE: MSG_GET_TREE_FMT,
+    MSG_CODE_TREE_DATA: MSG_TREE_DATA_FMT,
+    MSG_CODE_GET_DATA: MSG_GET_DATA_FMT,
+    MSG_CODE_NOTIFY: MSG_NOTIFY_FMT,
+}
+
+
+# Result formats
+MSG_FORMAT_XML = 1
+MSG_FORMAT_JSON = 2
+MSG_FORMAT_LYB = 3
+
+
+def cstr(mdata):
+    assert mdata[-1] == 0
+    return mdata[:-1]
+
+
+class FEClientError(Exception):
+    pass
+
+
+class PBMessageError(FEClientError):
+    def __init__(self, msg, errstr):
+        self.msg = msg
+        # self.sess_id = mhdr[HDR_FIELD_SESS_ID]
+        # self.req_id = mhdr[HDR_FIELD_REQ_ID]
+        self.error = -1
+        self.errstr = errstr
+        super().__init__(f"PBMessageError: {self.errstr}: {msg}")
+
+
+class NativeMessageError(FEClientError):
+    def __init__(self, mhdr, mfixed, mdata):
+        self.mhdr = mhdr
+        self.sess_id = mhdr[HDR_FIELD_SESS_ID]
+        self.req_id = mhdr[HDR_FIELD_REQ_ID]
+        self.error = mfixed[0]
+        self.errstr = cstr(mdata)
+        super().__init__(
+            "NativeMessageError: "
+            f"session {self.sess_id} reqid {self.req_id} "
+            f"error {self.error}: {self.errstr}"
+        )
+
+
+#
+# Low-level socket functions
+#
+
+
+def recv_wait(sock, size):
+    """Receive a fixed number of bytes from a stream socket."""
+    data = b""
+    while len(data) < size:
+        newdata = sock.recv(size - len(data))
+        if not newdata:
+            raise Exception("Socket closed")
+        data += newdata
+    return data
+
+
+def recv_msg(sock):
+    marker = recv_wait(sock, 4)
+    assert marker in (MGMT_MSG_MARKER_PROTOBUF, MGMT_MSG_MARKER_NATIVE)
+
+    msize = int.from_bytes(recv_wait(sock, 4), byteorder=sys.byteorder)
+    assert msize >= 8
+    mdata = recv_wait(sock, msize - 8) if msize > 8 else b""
+
+    return mdata, marker == MGMT_MSG_MARKER_NATIVE
+
+
+def send_msg(sock, marker, mdata):
+    """Send a mgmtd native message to a stream socket."""
+    msize = int.to_bytes(len(mdata) + 8, byteorder=sys.byteorder, length=4)
+    sock.send(marker)
+    sock.send(msize)
+    sock.send(mdata)
+
+
+class Session:
+    """A session to the mgmtd server."""
+
+    client_id = 1
+
+    def __init__(self, sock):
+        self.sock = sock
+        self.next_req_id = 1
+
+        req = mgmt_pb2.FeMessage()
+        req.register_req.client_name = "test-client"
+        self.send_pb_msg(req)
+        logging.debug("Sent FeRegisterReq: %s", req)
+
+        req = mgmt_pb2.FeMessage()
+        req.session_req.create = 1
+        req.session_req.client_conn_id = Session.client_id
+        Session.client_id += 1
+        self.send_pb_msg(req)
+        logging.debug("Sent FeSessionReq: %s", req)
+
+        reply = self.recv_pb_msg(mgmt_pb2.FeMessage())
+        logging.debug("Received FeSessionReply: %s", repr(reply))
+
+        assert reply.session_reply.success
+        self.sess_id = reply.session_reply.session_id
+
+    def close(self, clean=True):
+        if clean:
+            req = mgmt_pb2.FeMessage()
+            req.session_req.create = 0
+            req.session_req.sess_id = self.sess_id
+            self.send_pb_msg(req)
+        self.sock.close()
+        self.sock = None
+
+    def get_next_req_id(self):
+        req_id = self.next_req_id
+        self.next_req_id += 1
+        return req_id
+
+    # --------------------------
+    # Protobuf message functions
+    # --------------------------
+
+    def recv_pb_msg(self, msg):
+        """Receive a protobuf message."""
+        mdata, native = recv_msg(self.sock)
+        assert not native
+
+        msg.ParseFromString(mdata)
+
+        req = getattr(msg, msg.WhichOneof("message"))
+        if req.HasField("success"):
+            if not req.success:
+                raise PBMessageError(msg, req.error_if_any)
+
+        return msg
+
+    def send_pb_msg(self, msg):
+        """Send a protobuf message."""
+        mdata = msg.SerializeToString()
+        return send_msg(self.sock, MGMT_MSG_MARKER_PROTOBUF, mdata)
+
+    # ------------------------
+    # Native message functions
+    # ------------------------
+
+    def recv_native_msg(self):
+        """Send a native message."""
+        mdata, native = recv_msg(self.sock)
+        assert native
+
+        hlen = struct.calcsize(MSG_HDR_FMT)
+        hdata = mdata[:hlen]
+        mhdr = struct.unpack(MSG_HDR_FMT, hdata)
+        code = mhdr[0]
+
+        if code not in msg_native_formats:
+            raise Exception(f"Unknown native msg code {code} rcvd")
+
+        mfmt = msg_native_formats[code]
+        flen = struct.calcsize(mfmt)
+        fdata = mdata[hlen : hlen + flen]
+        mfixed = struct.unpack(mfmt, fdata)
+        mdata = mdata[hlen + flen :]
+
+        if code == MSG_ERROR_FMT:
+            raise NativeMessageError(mhdr, mfixed, mdata)
+
+        return mhdr, mfixed, mdata
+
+    def send_native_msg(self, mdata):
+        """Send a native message."""
+        return send_msg(self.sock, MGMT_MSG_MARKER_NATIVE, mdata)
+
+    def get_native_msg_header(self, msg_code):
+        req_id = self.get_next_req_id()
+        hdata = struct.pack(MSG_HDR_FMT, msg_code, 0, self.sess_id, req_id)
+        return hdata, req_id
+
+    # -----------------------
+    # Front-end API Fountains
+    # -----------------------
+
+    def lock(self, lock=True, ds_id=mgmt_pb2.CANDIDATE_DS):
+        req = mgmt_pb2.FeMessage()
+        req.lockds_req.session_id = self.sess_id
+        req.lockds_req.req_id = self.get_next_req_id()
+        req.lockds_req.ds_id = ds_id
+        req.lockds_req.lock = lock
+        self.send_pb_msg(req)
+        logging.debug("Sent LockDsReq: %s", req)
+
+        reply = self.recv_pb_msg(mgmt_pb2.FeMessage())
+        logging.debug("Received Reply: %s", repr(reply))
+        assert reply.lockds_reply.success
+
+    def get_data(self, query, data=True, config=False):
+        # Create the message
+        mdata, req_id = self.get_native_msg_header(MSG_CODE_GET_DATA)
+        flags = GET_DATA_FLAG_STATE if data else 0
+        flags |= GET_DATA_FLAG_CONFIG if config else 0
+        mdata += struct.pack(MSG_GET_DATA_FMT, MSG_FORMAT_JSON, flags)
+        mdata += query.encode("utf-8") + b"\x00"
+
+        self.send_native_msg(mdata)
+        logging.debug("Sent GET-TREE")
+
+        mhdr, mfixed, mdata = self.recv_native_msg()
+        assert mdata[-1] == 0
+        result = mdata[:-1].decode("utf-8")
+
+        logging.debug("Received GET: %s: %s", mfixed, mdata)
+        return result
+
+    # def subscribe(self, notif_xpath):
+    #     # Create the message
+    #     mdata, req_id = self.get_native_msg_header(MSG_CODE_SUBSCRIBE)
+    #     mdata += struct.pack(MSG_SUBSCRIBE_FMT, MSG_FORMAT_JSON)
+    #     mdata += notif_xpath.encode("utf-8") + b"\x00"
+
+    #     self.send_native_msg(mdata)
+    #     logging.debug("Sent SUBSCRIBE")
+
+    def recv_notify(self, xpaths=None):
+        while True:
+            logging.debug("Waiting for Notify Message")
+            mhdr, mfixed, mdata = self.recv_native_msg()
+            assert mdata[-1] == 0
+            result = mdata[:-1].decode("utf-8")
+            if mhdr[HDR_FIELD_CODE] == MSG_CODE_NOTIFY:
+                logging.debug("Received Notify Message: %s: %s", mfixed, mdata)
+            else:
+                raise Exception(f"Received NON-NOTIFY Message: {mfixed}: {mdata}")
+            if not xpaths:
+                return result
+            js = json.loads(result)
+            key = [x for x in js.keys()][0]
+            for xpath in xpaths:
+                if key.startswith(xpath):
+                    return result
+            logging.debug("'%s' didn't match xpath filters", key)
+
+
+def __parse_args():
+    MPATH = "/var/run/frr/mgmtd_fe.sock"
+    parser = argparse.ArgumentParser()
+    parser.add_argument(
+        "-l", "--listen", nargs="*", metavar="XPATH", help="xpath[s] to listen for"
+    )
+    parser.add_argument(
+        "--notify-count",
+        type=int,
+        default=1,
+        help="Number of notifications to listen for 0 for infinite",
+    )
+    parser.add_argument(
+        "-b", "--both", action="store_true", help="return both config and data"
+    )
+    parser.add_argument(
+        "-c", "--config-only", action="store_true", help="return config only"
+    )
+    parser.add_argument(
+        "-q", "--query", nargs="+", metavar="XPATH", help="xpath[s] to query"
+    )
+    parser.add_argument("-s", "--server", default=MPATH, help="path to server socket")
+    parser.add_argument("-v", "--verbose", action="store_true", help="Be verbose")
+    args = parser.parse_args()
+
+    level = logging.DEBUG if args.verbose else logging.INFO
+    logging.basicConfig(level=level, format="%(asctime)s %(levelname)s: %(message)s")
+
+    return args
+
+
+def __server_connect(spath):
+    sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
+    logging.debug("Connecting to server on %s", spath)
+    while ec := sock.connect_ex(str(spath)):
+        logging.warn("retry server connection in .5s (%s)", os.strerror(ec))
+        time.sleep(0.5)
+    logging.info("Connected to server on %s", spath)
+    return sock
+
+
+def __main():
+    args = __parse_args()
+    sock = __server_connect(Path(args.server))
+    sess = Session(sock)
+
+    if args.query:
+        # Performa an xpath query
+        # query = "/frr-interface:lib/interface/state/mtu"
+        for query in args.query:
+            logging.info("Sending query: %s", query)
+            result = sess.get_data(
+                query, data=not args.config_only, config=(args.both or args.config_only)
+            )
+            print(result)
+
+    if args.listen is not None:
+        i = args.notify_count
+        while i > 0 or args.notify_count == 0:
+            notif = sess.recv_notify(args.listen)
+            print(notif)
+            i -= 1
+
+
+def main():
+    try:
+        __main()
+    except KeyboardInterrupt:
+        logging.info("Exiting")
+    except Exception as error:
+        logging.error("Unexpected error exiting: %s", error, exc_info=True)
+
+
+if __name__ == "__main__":
+    main()

tests/topotests/mgmt_fe_client/fe_client.py

@@ -1,103 +0,0 @@
-#!/usr/bin/env python
-# -*- coding: utf-8 eval: (blacken-mode 1) -*-
-# SPDX-License-Identifier: GPL-2.0-or-later
-#
-# November 27 2023, Christian Hopps <chopps@labn.net>
-#
-# Copyright (c) 2023, LabN Consulting, L.L.C.
-#
-# noqa: E501
-#
-import argparse
-import errno
-import logging
-import os
-import socket
-import sys
-import time
-from pathlib import Path
-
-import mgmt_pb2
-
-MGMT_MSG_MARKER_PROTOBUF = b"\000###"
-MGMT_MSG_MARKER_NATIVE = b"\001###"
-
-
-def __parse_args():
-    MPATH = "/var/run/frr/mgmtd_fe.sock"
-    parser = argparse.ArgumentParser()
-    parser.add_argument("--verbose", action="store_true", help="Be verbose")
-    parser.add_argument("--server", default=MPATH, help="path to server socket")
-    args = parser.parse_args()
-
-    level = logging.DEBUG if args.verbose else logging.INFO
-    logging.basicConfig(level=level, format="%(asctime)s %(levelname)s: %(message)s")
-
-    return args
-
-
-def __server_connect(spath):
-    sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
-    logging.debug("Connecting to server on %s", spath)
-    while ec := sock.connect_ex(str(spath)):
-        logging.warn("retry server connection in .5s (%s)", os.strerror(ec))
-        time.sleep(0.5)
-    logging.info("Connected to server on %s", spath)
-    return sock
-
-
-def mgmt_pb_recv_msg(sock, msg):
-    """Receive a mgmtd protobuf message from a stream socket."""
-    marker = sock.recv(4)
-    assert marker in (MGMT_MSG_MARKER_PROTOBUF, MGMT_MSG_MARKER_NATIVE)
-
-    msize = int.from_bytes(sock.recv(4), byteorder="big")
-    mdata = sock.recv(msize)
-
-    msg.ParseFromString(mdata)
-    return msg
-
-
-def mgmt_pb_send_msg(sock, msg):
-    """Send a mgmtd protobuf message from a stream socket."""
-    marker = MGMT_MSG_MARKER_PROTOBUF
-    mdata = msg.SerializeToString()
-    msize = int.to_bytes(len(mdata), byteorder="big", length=4)
-    sock.send(marker)
-    sock.send(msize)
-    sock.send(mdata)
-
-
-def create_session(sock):
-    req = mgmt_pb2.FeRegisterReq()
-    req.client_name = "test-client"
-    mgmt_pb_send_msg(sock, req)
-    logging.debug("Sent FeRegisterReq: %s", req)
-
-    req = mgmt_pb2.FeSessionReq()
-    req.create = 1
-    req.client_conn_id = 1
-    mgmt_pb_send_msg(sock, req)
-    logging.debug("Sent FeSessionReq: %s", req)
-
-    reply = mgmt_pb_recv_msg(sock, mgmt_pb2.FeSessionReply())
-    logging.debug("Received FeSessionReply: %s", reply)
-
-
-def __main():
-    args = __parse_args()
-    sock = __server_connect(Path(args.server))
-    create_session(sock)
-
-
-def main():
-    try:
-        __main()
-    except KeyboardInterrupt:
-        logging.info("Exiting")
-    except Exception as error:
-        logging.error("Unexpected error exiting: %s", error, exc_info=True)
-
-
-if __name__ == "__main__":
-    main()

tests/topotests/mgmt_fe_client/test_client.py

@@ -1,4 +1,3 @@
-#!/usr/bin/env python
 # -*- coding: utf-8 eval: (blacken-mode 1) -*-
 # SPDX-License-Identifier: ISC
 #

tests/topotests/mgmt_notif/oper.py

@@ -0,0 +1 @@
+../mgmt_oper/oper.py

tests/topotests/mgmt_notif/r1/frr.conf

@@ -0,0 +1,27 @@
+log timestamp precision 6
+log file frr.log
+
+no debug memstats-at-exit
+
+debug northbound notifications
+debug northbound libyang
+debug northbound events
+debug northbound callbacks
+
+debug mgmt backend datastore frontend transaction
+debug mgmt client frontend
+debug mgmt client backend
+
+ip route 11.11.11.11/32 lo
+
+interface r1-eth0
+ ip address 1.1.1.1/24
+ ip rip authentication string foo
+ ip rip authentication mode text
+exit
+
+router rip
+ network 1.1.1.0/24
+ timers basic 5 15 10
+ redistribute static
+exit

tests/topotests/mgmt_notif/r2/frr.conf

@@ -0,0 +1,27 @@
+log timestamp precision 6
+log file frr.log
+
+no debug memstats-at-exit
+
+debug northbound notifications
+debug northbound libyang
+debug northbound events
+debug northbound callbacks
+
+debug mgmt backend datastore frontend transaction
+debug mgmt client frontend
+debug mgmt client backend
+
+ip route 22.22.22.22/32 lo
+
+interface r2-eth0
+ ip address 1.1.1.2/24
+ ip rip authentication string bar
+ ip rip authentication mode text
+exit
+
+router rip
+ network 1.1.1.0/24
+ timers basic 5 15 10
+ redistribute static
+exit

tests/topotests/mgmt_notif/test_notif.py

@@ -0,0 +1,65 @@
+# -*- coding: utf-8 eval: (blacken-mode 1) -*-
+# SPDX-License-Identifier: ISC
+#
+# January 23 2024, Christian Hopps <chopps@labn.net>
+#
+# Copyright (c) 2024, LabN Consulting, L.L.C.
+#
+
+"""
+Test YANG Notifications
+"""
+import json
+import logging
+import os
+
+import pytest
+from lib.topogen import Topogen
+from lib.topotest import json_cmp
+from oper import check_kernel_32
+
+pytestmark = [pytest.mark.ripd, pytest.mark.staticd, pytest.mark.mgmtd]
+
+CWD = os.path.dirname(os.path.realpath(__file__))
+
+
+@pytest.fixture(scope="module")
+def tgen(request):
+    "Setup/Teardown the environment and provide tgen argument to tests"
+
+    topodef = {
+        "s1": ("r1", "r2"),
+    }
+
+    tgen = Topogen(topodef, request.module.__name__)
+    tgen.start_topology()
+
+    router_list = tgen.routers()
+    for rname, router in router_list.items():
+        router.load_frr_config("frr.conf")
+
+    tgen.start_router()
+    yield tgen
+    tgen.stop_topology()
+
+
+def test_oper_simple(tgen):
+    if tgen.routers_have_failure():
+        pytest.skip(tgen.errors)
+
+    r1 = tgen.gears["r1"].net
+
+    check_kernel_32(r1, "11.11.11.11", 1, "")
+
+    fe_client_path = CWD + "/../lib/fe_client.py"
+    rc, _, _ = r1.cmd_status(fe_client_path + " --help")
+
+    if rc:
+        pytest.skip("No protoc or present cannot run test")
+
+    output = r1.cmd_raises(fe_client_path + " --listen")
+    jsout = json.loads(output)
+
+    expected = {"frr-ripd:authentication-type-failure": {"interface-name": "r1-eth0"}}
+    result = json_cmp(jsout, expected)
+    assert result is None