1 files changed, 275 insertions, 86 deletions

diff --git a/lib/zclient.c b/lib/zclient.c
index eac6c7081d..be2c4e54a0 100644
--- a/lib/zclient.c
+++ b/lib/zclient.c
@@ -52,7 +52,8 @@ static void zclient_event(enum event, struct zclient *);
 static void zebra_interface_if_set_value(struct stream *s,
 					 struct interface *ifp);
 
-struct zclient_options zclient_options_default = {.receive_notify = false};
+struct zclient_options zclient_options_default = {.receive_notify = false,
+						  .synchronous = false};
 
 struct sockaddr_storage zclient_addr;
 socklen_t zclient_addr_len;
@@ -76,6 +77,7 @@ struct zclient *zclient_new(struct thread_master *master,
 	zclient->master = master;
 
 	zclient->receive_notify = opt->receive_notify;
+	zclient->synchronous = opt->synchronous;
 
 	return zclient;
 }
@@ -374,11 +376,11 @@ static int zebra_message_send(struct zclient *zclient, int command,
 	return zclient_send_message(zclient);
 }
 
-static int zebra_hello_send(struct zclient *zclient)
+int zclient_send_hello(struct zclient *zclient)
 {
 	struct stream *s;
 
-	if (zclient->redist_default) {
+	if (zclient->redist_default || zclient->synchronous) {
 		s = zclient->obuf;
 		stream_reset(s);
 
@@ -386,10 +388,15 @@ static int zebra_hello_send(struct zclient *zclient)
 		zclient_create_header(s, ZEBRA_HELLO, VRF_DEFAULT);
 		stream_putc(s, zclient->redist_default);
 		stream_putw(s, zclient->instance);
+		stream_putl(s, zclient->session_id);
 		if (zclient->receive_notify)
 			stream_putc(s, 1);
 		else
 			stream_putc(s, 0);
+		if (zclient->synchronous)
+			stream_putc(s, 1);
+		else
+			stream_putc(s, 0);
 
 		stream_putw_at(s, 0, stream_get_endp(s));
 		return zclient_send_message(zclient);
@@ -629,7 +636,7 @@ int zclient_start(struct zclient *zclient)
 	/* Create read thread. */
 	zclient_event(ZCLIENT_READ, zclient);
 
-	zebra_hello_send(zclient);
+	zclient_send_hello(zclient);
 
 	zebra_message_send(zclient, ZEBRA_INTERFACE_ADD, VRF_DEFAULT);
 
@@ -690,8 +697,9 @@ static int zclient_connect(struct thread *t)
 	return zclient_start(zclient);
 }
 
-int zclient_send_rnh(struct zclient *zclient, int command, struct prefix *p,
-		     bool exact_match, vrf_id_t vrf_id)
+int zclient_send_rnh(struct zclient *zclient, int command,
+		     const struct prefix *p, bool exact_match,
+		     vrf_id_t vrf_id)
 {
 	struct stream *s;
 
@@ -897,6 +905,7 @@ int zapi_nexthop_encode(struct stream *s, const struct zapi_nexthop *api_nh,
 		}
 	}
 
+	/* If present, set 'weight' flag before encoding flags */
 	if (api_nh->weight)
 		SET_FLAG(nh_flags, ZAPI_NEXTHOP_FLAG_WEIGHT);
 
@@ -941,6 +950,10 @@ int zapi_nexthop_encode(struct stream *s, const struct zapi_nexthop *api_nh,
 		stream_put(s, &(api_nh->rmac),
 			   sizeof(struct ethaddr));
 
+	/* Index of backup nexthop */
+	if (CHECK_FLAG(nh_flags, ZAPI_NEXTHOP_FLAG_HAS_BACKUP))
+		stream_putc(s, api_nh->backup_idx);
+
 done:
 	return ret;
 }
@@ -978,7 +991,7 @@ int zapi_route_encode(uint8_t cmd, struct stream *s, struct zapi_route *api)
 	stream_putc(s, api->prefix.family);
 	psize = PSIZE(api->prefix.prefixlen);
 	stream_putc(s, api->prefix.prefixlen);
-	stream_write(s, (uint8_t *)&api->prefix.u.prefix, psize);
+	stream_write(s, &api->prefix.u.prefix, psize);
 
 	if (CHECK_FLAG(api->message, ZAPI_MESSAGE_SRCPFX)) {
 		psize = PSIZE(api->src_prefix.prefixlen);
@@ -1000,6 +1013,10 @@ int zapi_route_encode(uint8_t cmd, struct stream *s, struct zapi_route *api)
 			return -1;
 		}
 
+		/* We canonicalize the nexthops by sorting them; this allows
+		 * zebra to resolve the list of nexthops to a nexthop-group
+		 * more efficiently.
+		 */
 		zapi_nexthop_group_sort(api->nexthops, api->nexthop_num);
 
 		stream_putw(s, api->nexthop_num);
@@ -1026,6 +1043,50 @@ int zapi_route_encode(uint8_t cmd, struct stream *s, struct zapi_route *api)
 		}
 	}
 
+	/* Backup nexthops  */
+	if (CHECK_FLAG(api->message, ZAPI_MESSAGE_BACKUP_NEXTHOPS)) {
+		/* limit the number of nexthops if necessary */
+		if (api->backup_nexthop_num > MULTIPATH_NUM) {
+			char buf[PREFIX2STR_BUFFER];
+
+			prefix2str(&api->prefix, buf, sizeof(buf));
+			flog_err(
+				EC_LIB_ZAPI_ENCODE,
+				"%s: prefix %s: can't encode %u backup nexthops (maximum is %u)",
+				__func__, buf, api->backup_nexthop_num,
+				MULTIPATH_NUM);
+			return -1;
+		}
+
+		/* Note that we do not sort the list of backup nexthops -
+		 * this list is treated as an array and indexed by each
+		 * primary nexthop that is associated with a backup.
+		 */
+
+		stream_putw(s, api->backup_nexthop_num);
+
+		for (i = 0; i < api->backup_nexthop_num; i++) {
+			api_nh = &api->backup_nexthops[i];
+
+			/* MPLS labels for BGP-LU or Segment Routing */
+			if (api_nh->label_num > MPLS_MAX_LABELS) {
+				char buf[PREFIX2STR_BUFFER];
+
+				prefix2str(&api->prefix, buf, sizeof(buf));
+
+				flog_err(EC_LIB_ZAPI_ENCODE,
+					 "%s: prefix %s: backup: can't encode %u labels (maximum is %u)",
+					 __func__, buf,
+					 api_nh->label_num,
+					 MPLS_MAX_LABELS);
+				return -1;
+			}
+
+			if (zapi_nexthop_encode(s, api_nh, api->flags) != 0)
+				return -1;
+		}
+	}
+
 	/* Attributes. */
 	if (CHECK_FLAG(api->message, ZAPI_MESSAGE_DISTANCE))
 		stream_putc(s, api->distance);
@@ -1101,6 +1162,10 @@ static int zapi_nexthop_decode(struct stream *s, struct zapi_nexthop *api_nh,
 		STREAM_GET(&(api_nh->rmac), s,
 			   sizeof(struct ethaddr));
 
+	/* Backup nexthop index */
+	if (CHECK_FLAG(api_nh->flags, ZAPI_NEXTHOP_FLAG_HAS_BACKUP))
+		STREAM_GETC(s, api_nh->backup_idx);
+
 	/* Success */
 	ret = 0;
 
@@ -1207,6 +1272,24 @@ int zapi_route_decode(struct stream *s, struct zapi_route *api)
 		}
 	}
 
+	/* Backup nexthops. */
+	if (CHECK_FLAG(api->message, ZAPI_MESSAGE_BACKUP_NEXTHOPS)) {
+		STREAM_GETW(s, api->backup_nexthop_num);
+		if (api->backup_nexthop_num > MULTIPATH_NUM) {
+			flog_err(EC_LIB_ZAPI_ENCODE,
+				 "%s: invalid number of backup nexthops (%u)",
+				 __func__, api->backup_nexthop_num);
+			return -1;
+		}
+
+		for (i = 0; i < api->backup_nexthop_num; i++) {
+			api_nh = &api->backup_nexthops[i];
+
+			if (zapi_nexthop_decode(s, api_nh, api->flags) != 0)
+				return -1;
+		}
+	}
+
 	/* Attributes. */
 	if (CHECK_FLAG(api->message, ZAPI_MESSAGE_DISTANCE))
 		STREAM_GETC(s, api->distance);
@@ -1381,7 +1464,7 @@ stream_failure:
 	return false;
 }
 
-struct nexthop *nexthop_from_zapi_nexthop(struct zapi_nexthop *znh)
+struct nexthop *nexthop_from_zapi_nexthop(const struct zapi_nexthop *znh)
 {
 	struct nexthop *n = nexthop_new();
 
@@ -1398,6 +1481,11 @@ struct nexthop *nexthop_from_zapi_nexthop(struct zapi_nexthop *znh)
 				   znh->labels);
 	}
 
+	if (CHECK_FLAG(znh->flags, ZAPI_NEXTHOP_FLAG_HAS_BACKUP)) {
+		SET_FLAG(n->flags, NEXTHOP_FLAG_HAS_BACKUP);
+		n->backup_idx = znh->backup_idx;
+	}
+
 	return n;
 }
 
@@ -1413,10 +1501,16 @@ int zapi_nexthop_from_nexthop(struct zapi_nexthop *znh,
 
 	znh->type = nh->type;
 	znh->vrf_id = nh->vrf_id;
+	znh->weight = nh->weight;
 	znh->ifindex = nh->ifindex;
 	znh->gate = nh->gate;
 
 	if (nh->nh_label && (nh->nh_label->num_labels > 0)) {
+
+		/* Validate */
+		if (nh->nh_label->num_labels > MPLS_MAX_LABELS)
+			return -1;
+
 		for (i = 0; i < nh->nh_label->num_labels; i++)
 			znh->labels[i] = nh->nh_label->label[i];
 
@@ -1424,10 +1518,31 @@ int zapi_nexthop_from_nexthop(struct zapi_nexthop *znh,
 		SET_FLAG(znh->flags, ZAPI_NEXTHOP_FLAG_LABEL);
 	}
 
+	if (CHECK_FLAG(nh->flags, NEXTHOP_FLAG_HAS_BACKUP)) {
+		SET_FLAG(znh->flags, ZAPI_NEXTHOP_FLAG_HAS_BACKUP);
+		znh->backup_idx = nh->backup_idx;
+	}
+
 	return 0;
 }
 
 /*
+ * Wrapper that converts backup nexthop
+ */
+int zapi_backup_nexthop_from_nexthop(struct zapi_nexthop *znh,
+				     const struct nexthop *nh)
+{
+	int ret;
+
+	/* Ensure that zapi flags are correct: backups don't have backups */
+	ret = zapi_nexthop_from_nexthop(znh, nh);
+	if (ret == 0)
+		UNSET_FLAG(znh->flags, ZAPI_NEXTHOP_FLAG_HAS_BACKUP);
+
+	return ret;
+}
+
+/*
  * Decode the nexthop-tracking update message
  */
 bool zapi_nexthop_update_decode(struct stream *s, struct zapi_route *nhr)
@@ -1521,33 +1636,34 @@ int zebra_redistribute_default_send(int command, struct zclient *zclient,
 }
 
 /* Get prefix in ZServ format; family should be filled in on prefix */
-static void zclient_stream_get_prefix(struct stream *s, struct prefix *p)
+static int zclient_stream_get_prefix(struct stream *s, struct prefix *p)
 {
 	size_t plen = prefix_blen(p);
 	uint8_t c;
 	p->prefixlen = 0;
 
 	if (plen == 0)
-		return;
+		return -1;
 
-	stream_get(&p->u.prefix, s, plen);
+	STREAM_GET(&p->u.prefix, s, plen);
 	STREAM_GETC(s, c);
 	p->prefixlen = MIN(plen * 8, c);
 
+	return 0;
 stream_failure:
-	return;
+	return -1;
 }
 
 /* Router-id update from zebra daemon. */
-void zebra_router_id_update_read(struct stream *s, struct prefix *rid)
+int zebra_router_id_update_read(struct stream *s, struct prefix *rid)
 {
 	/* Fetch interface address. */
 	STREAM_GETC(s, rid->family);
 
-	zclient_stream_get_prefix(s, rid);
+	return zclient_stream_get_prefix(s, rid);
 
 stream_failure:
-	return;
+	return -1;
 }
 
 /* Interface addition from zebra daemon. */
@@ -1596,24 +1712,36 @@ stream_failure:
  * +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
  */
 
-static void zclient_vrf_add(struct zclient *zclient, vrf_id_t vrf_id)
+static int zclient_vrf_add(struct zclient *zclient, vrf_id_t vrf_id)
 {
 	struct vrf *vrf;
-	char vrfname_tmp[VRF_NAMSIZ];
+	char vrfname_tmp[VRF_NAMSIZ + 1] = {};
 	struct vrf_data data;
 
-	stream_get(&data, zclient->ibuf, sizeof(struct vrf_data));
+	STREAM_GET(&data, zclient->ibuf, sizeof(struct vrf_data));
 	/* Read interface name. */
-	stream_get(vrfname_tmp, zclient->ibuf, VRF_NAMSIZ);
+	STREAM_GET(vrfname_tmp, zclient->ibuf, VRF_NAMSIZ);
+
+	if (strlen(vrfname_tmp) == 0)
+		goto stream_failure;
 
-	/* Lookup/create vrf by vrf_id. */
+	/* Lookup/create vrf by name, then vrf_id. */
 	vrf = vrf_get(vrf_id, vrfname_tmp);
+
+	/* If there's already a VRF with this name, don't create vrf */
+	if (!vrf)
+		return 0;
+
 	vrf->data.l.table_id = data.l.table_id;
 	memcpy(vrf->data.l.netns_name, data.l.netns_name, NS_NAMSIZ);
 	/* overwrite default vrf */
 	if (vrf_id == VRF_DEFAULT)
 		vrf_set_default_name(vrfname_tmp, false);
 	vrf_enable(vrf);
+
+	return 0;
+stream_failure:
+	return -1;
 }
 
 static void zclient_vrf_delete(struct zclient *zclient, vrf_id_t vrf_id)
@@ -1634,21 +1762,32 @@ static void zclient_vrf_delete(struct zclient *zclient, vrf_id_t vrf_id)
 	vrf_delete(vrf);
 }
 
-static void zclient_interface_add(struct zclient *zclient, vrf_id_t vrf_id)
+static int zclient_interface_add(struct zclient *zclient, vrf_id_t vrf_id)
 {
 	struct interface *ifp;
-	char ifname_tmp[INTERFACE_NAMSIZ];
+	char ifname_tmp[INTERFACE_NAMSIZ + 1] = {};
 	struct stream *s = zclient->ibuf;
 
 	/* Read interface name. */
-	stream_get(ifname_tmp, s, INTERFACE_NAMSIZ);
+	STREAM_GET(ifname_tmp, s, INTERFACE_NAMSIZ);
 
 	/* Lookup/create interface by name. */
+	if (!vrf_get(vrf_id, NULL)) {
+		zlog_debug(
+			"Rx'd interface add from Zebra, but VRF %u does not exist",
+			vrf_id);
+		return -1;
+	}
+
 	ifp = if_get_by_name(ifname_tmp, vrf_id);
 
 	zebra_interface_if_set_value(s, ifp);
 
 	if_new_via_zapi(ifp);
+
+	return 0;
+stream_failure:
+	return -1;
 }
 
 /*
@@ -1660,10 +1799,10 @@ static void zclient_interface_add(struct zclient *zclient, vrf_id_t vrf_id)
 struct interface *zebra_interface_state_read(struct stream *s, vrf_id_t vrf_id)
 {
 	struct interface *ifp;
-	char ifname_tmp[INTERFACE_NAMSIZ];
+	char ifname_tmp[INTERFACE_NAMSIZ + 1] = {};
 
 	/* Read interface name. */
-	stream_get(ifname_tmp, s, INTERFACE_NAMSIZ);
+	STREAM_GET(ifname_tmp, s, INTERFACE_NAMSIZ);
 
 	/* Lookup this by interface index. */
 	ifp = if_lookup_by_name(ifname_tmp, vrf_id);
@@ -1677,6 +1816,8 @@ struct interface *zebra_interface_state_read(struct stream *s, vrf_id_t vrf_id)
 	zebra_interface_if_set_value(s, ifp);
 
 	return ifp;
+stream_failure:
+	return NULL;
 }
 
 static void zclient_interface_delete(struct zclient *zclient, vrf_id_t vrf_id)
@@ -1730,21 +1871,23 @@ static void zclient_handle_error(ZAPI_CALLBACK_ARGS)
 		(*zclient->handle_error)(error);
 }
 
-static void link_params_set_value(struct stream *s, struct if_link_params *iflp)
+static int link_params_set_value(struct stream *s, struct if_link_params *iflp)
 {
 
 	if (iflp == NULL)
-		return;
+		return -1;
 
-	iflp->lp_status = stream_getl(s);
-	iflp->te_metric = stream_getl(s);
-	iflp->max_bw = stream_getf(s);
-	iflp->max_rsv_bw = stream_getf(s);
-	uint32_t bwclassnum = stream_getl(s);
+	uint32_t bwclassnum;
+
+	STREAM_GETL(s, iflp->lp_status);
+	STREAM_GETL(s, iflp->te_metric);
+	STREAM_GETF(s, iflp->max_bw);
+	STREAM_GETF(s, iflp->max_rsv_bw);
+	STREAM_GETL(s, bwclassnum);
 	{
 		unsigned int i;
 		for (i = 0; i < bwclassnum && i < MAX_CLASS_TYPE; i++)
-			iflp->unrsv_bw[i] = stream_getf(s);
+			STREAM_GETF(s, iflp->unrsv_bw[i]);
 		if (i < bwclassnum)
 			flog_err(
 				EC_LIB_ZAPI_MISSMATCH,
@@ -1752,19 +1895,23 @@ static void link_params_set_value(struct stream *s, struct if_link_params *iflp)
 				" - outdated library?",
 				__func__, bwclassnum, MAX_CLASS_TYPE);
 	}
-	iflp->admin_grp = stream_getl(s);
-	iflp->rmt_as = stream_getl(s);
+	STREAM_GETL(s, iflp->admin_grp);
+	STREAM_GETL(s, iflp->rmt_as);
 	iflp->rmt_ip.s_addr = stream_get_ipv4(s);
 
-	iflp->av_delay = stream_getl(s);
-	iflp->min_delay = stream_getl(s);
-	iflp->max_delay = stream_getl(s);
-	iflp->delay_var = stream_getl(s);
+	STREAM_GETL(s, iflp->av_delay);
+	STREAM_GETL(s, iflp->min_delay);
+	STREAM_GETL(s, iflp->max_delay);
+	STREAM_GETL(s, iflp->delay_var);
+
+	STREAM_GETF(s, iflp->pkt_loss);
+	STREAM_GETF(s, iflp->res_bw);
+	STREAM_GETF(s, iflp->ava_bw);
+	STREAM_GETF(s, iflp->use_bw);
 
-	iflp->pkt_loss = stream_getf(s);
-	iflp->res_bw = stream_getf(s);
-	iflp->ava_bw = stream_getf(s);
-	iflp->use_bw = stream_getf(s);
+	return 0;
+stream_failure:
+	return -1;
 }
 
 struct interface *zebra_interface_link_params_read(struct stream *s,
@@ -1773,9 +1920,7 @@ struct interface *zebra_interface_link_params_read(struct stream *s,
 	struct if_link_params *iflp;
 	ifindex_t ifindex;
 
-	assert(s);
-
-	ifindex = stream_getl(s);
+	STREAM_GETL(s, ifindex);
 
 	struct interface *ifp = if_lookup_by_index(ifindex, vrf_id);
 
@@ -1789,36 +1934,41 @@ struct interface *zebra_interface_link_params_read(struct stream *s,
 	if ((iflp = if_link_params_get(ifp)) == NULL)
 		return NULL;
 
-	link_params_set_value(s, iflp);
+	if (link_params_set_value(s, iflp) != 0)
+		goto stream_failure;
 
 	return ifp;
+
+stream_failure:
+	return NULL;
 }
 
 static void zebra_interface_if_set_value(struct stream *s,
 					 struct interface *ifp)
 {
 	uint8_t link_params_status = 0;
-	ifindex_t old_ifindex;
+	ifindex_t old_ifindex, new_ifindex;
 
 	old_ifindex = ifp->ifindex;
 	/* Read interface's index. */
-	if_set_index(ifp, stream_getl(s));
-	ifp->status = stream_getc(s);
+	STREAM_GETL(s, new_ifindex);
+	if_set_index(ifp, new_ifindex);
+	STREAM_GETC(s, ifp->status);
 
 	/* Read interface's value. */
-	ifp->flags = stream_getq(s);
-	ifp->ptm_enable = stream_getc(s);
-	ifp->ptm_status = stream_getc(s);
-	ifp->metric = stream_getl(s);
-	ifp->speed = stream_getl(s);
-	ifp->mtu = stream_getl(s);
-	ifp->mtu6 = stream_getl(s);
-	ifp->bandwidth = stream_getl(s);
-	ifp->link_ifindex = stream_getl(s);
-	ifp->ll_type = stream_getl(s);
-	ifp->hw_addr_len = stream_getl(s);
+	STREAM_GETQ(s, ifp->flags);
+	STREAM_GETC(s, ifp->ptm_enable);
+	STREAM_GETC(s, ifp->ptm_status);
+	STREAM_GETL(s, ifp->metric);
+	STREAM_GETL(s, ifp->speed);
+	STREAM_GETL(s, ifp->mtu);
+	STREAM_GETL(s, ifp->mtu6);
+	STREAM_GETL(s, ifp->bandwidth);
+	STREAM_GETL(s, ifp->link_ifindex);
+	STREAM_GETL(s, ifp->ll_type);
+	STREAM_GETL(s, ifp->hw_addr_len);
 	if (ifp->hw_addr_len)
-		stream_get(ifp->hw_addr, s,
+		STREAM_GET(ifp->hw_addr, s,
 			   MIN(ifp->hw_addr_len, INTERFACE_HWADDR_MAX));
 
 	/* Read Traffic Engineering status */
@@ -1830,6 +1980,11 @@ static void zebra_interface_if_set_value(struct stream *s,
 	}
 
 	nexthop_group_interface_state_change(ifp, old_ifindex);
+
+	return;
+stream_failure:
+	zlog_err("Could not parse interface values; aborting");
+	assert(!"Failed to parse interface values");
 }
 
 size_t zebra_interface_link_params_write(struct stream *s,
@@ -1928,7 +2083,7 @@ struct connected *zebra_interface_address_read(int type, struct stream *s,
 	memset(&d, 0, sizeof(d));
 
 	/* Get interface index. */
-	ifindex = stream_getl(s);
+	STREAM_GETL(s, ifindex);
 
 	/* Lookup index. */
 	ifp = if_lookup_by_index(ifindex, vrf_id);
@@ -1941,16 +2096,18 @@ struct connected *zebra_interface_address_read(int type, struct stream *s,
 	}
 
 	/* Fetch flag. */
-	ifc_flags = stream_getc(s);
+	STREAM_GETC(s, ifc_flags);
 
 	/* Fetch interface address. */
-	d.family = p.family = stream_getc(s);
+	STREAM_GETC(s, d.family);
+	p.family = d.family;
 	plen = prefix_blen(&d);
 
-	zclient_stream_get_prefix(s, &p);
+	if (zclient_stream_get_prefix(s, &p) != 0)
+		goto stream_failure;
 
 	/* Fetch destination address. */
-	stream_get(&d.u.prefix, s, plen);
+	STREAM_GET(&d.u.prefix, s, plen);
 
 	/* N.B. NULL destination pointers are encoded as all zeroes */
 	dp = memconstant(&d.u.prefix, 0, plen) ? NULL : &d;
@@ -1986,6 +2143,9 @@ struct connected *zebra_interface_address_read(int type, struct stream *s,
 	}
 
 	return ifc;
+
+stream_failure:
+	return NULL;
 }
 
 /*
@@ -2021,7 +2181,7 @@ zebra_interface_nbr_address_read(int type, struct stream *s, vrf_id_t vrf_id)
 	struct nbr_connected *ifc;
 
 	/* Get interface index. */
-	ifindex = stream_getl(s);
+	STREAM_GETL(s, ifindex);
 
 	/* Lookup index. */
 	ifp = if_lookup_by_index(ifindex, vrf_id);
@@ -2034,9 +2194,9 @@ zebra_interface_nbr_address_read(int type, struct stream *s, vrf_id_t vrf_id)
 		return NULL;
 	}
 
-	p.family = stream_getc(s);
-	stream_get(&p.u.prefix, s, prefix_blen(&p));
-	p.prefixlen = stream_getc(s);
+	STREAM_GETC(s, p.family);
+	STREAM_GET(&p.u.prefix, s, prefix_blen(&p));
+	STREAM_GETC(s, p.prefixlen);
 
 	if (type == ZEBRA_INTERFACE_NBR_ADDRESS_ADD) {
 		/* Currently only supporting P2P links, so any new RA source
@@ -2060,18 +2220,21 @@ zebra_interface_nbr_address_read(int type, struct stream *s, vrf_id_t vrf_id)
 	}
 
 	return ifc;
+
+stream_failure:
+	return NULL;
 }
 
 struct interface *zebra_interface_vrf_update_read(struct stream *s,
 						  vrf_id_t vrf_id,
 						  vrf_id_t *new_vrf_id)
 {
-	char ifname[INTERFACE_NAMSIZ];
+	char ifname[INTERFACE_NAMSIZ + 1] = {};
 	struct interface *ifp;
 	vrf_id_t new_id;
 
 	/* Read interface name. */
-	stream_get(ifname, s, INTERFACE_NAMSIZ);
+	STREAM_GET(ifname, s, INTERFACE_NAMSIZ);
 
 	/* Lookup interface. */
 	ifp = if_lookup_by_name(ifname, vrf_id);
@@ -2083,10 +2246,13 @@ struct interface *zebra_interface_vrf_update_read(struct stream *s,
 	}
 
 	/* Fetch new VRF Id. */
-	new_id = stream_getl(s);
+	STREAM_GETL(s, new_id);
 
 	*new_vrf_id = new_id;
 	return ifp;
+
+stream_failure:
+	return NULL;
 }
 
 /* filter unwanted messages until the expected one arrives */
@@ -2195,8 +2361,11 @@ int lm_label_manager_connect(struct zclient *zclient, int async)
 	s = zclient->ibuf;
 
 	/* read instance and proto */
-	uint8_t proto = stream_getc(s);
-	uint16_t instance = stream_getw(s);
+	uint8_t proto;
+	uint16_t instance;
+
+	STREAM_GETC(s, proto);
+	STREAM_GETW(s, instance);
 
 	/* sanity */
 	if (proto != zclient->redist_default)
@@ -2211,11 +2380,14 @@ int lm_label_manager_connect(struct zclient *zclient, int async)
 			instance, zclient->instance);
 
 	/* result code */
-	result = stream_getc(s);
+	STREAM_GETC(s, result);
 	if (zclient_debug)
 		zlog_debug("LM connect-response received, result %u", result);
 
 	return (int)result;
+
+stream_failure:
+	return -1;
 }
 
 /*
@@ -2323,8 +2495,11 @@ int lm_get_label_chunk(struct zclient *zclient, uint8_t keep, uint32_t base,
 	s = zclient->ibuf;
 
 	/* read proto and instance */
-	uint8_t proto = stream_getc(s);
-	uint16_t instance = stream_getw(s);
+	uint8_t proto;
+	uint8_t instance;
+
+	STREAM_GETC(s, proto);
+	STREAM_GETW(s, instance);
 
 	/* sanities */
 	if (proto != zclient->redist_default)
@@ -2346,10 +2521,10 @@ int lm_get_label_chunk(struct zclient *zclient, uint8_t keep, uint32_t base,
 	}
 
 	/* keep */
-	response_keep = stream_getc(s);
+	STREAM_GETC(s, response_keep);
 	/* start and end labels */
-	*start = stream_getl(s);
-	*end = stream_getl(s);
+	STREAM_GETL(s, *start);
+	STREAM_GETL(s, *end);
 
 	/* not owning this response */
 	if (keep != response_keep) {
@@ -2371,6 +2546,9 @@ int lm_get_label_chunk(struct zclient *zclient, uint8_t keep, uint32_t base,
 			   response_keep);
 
 	return 0;
+
+stream_failure:
+	return -1;
 }
 
 /**
@@ -2760,7 +2938,7 @@ int zebra_send_pw(struct zclient *zclient, int command, struct zapi_pw *pw)
 /*
  * Receive PW status update from Zebra and send it to LDE process.
  */
-void zebra_read_pw_status_update(ZAPI_CALLBACK_ARGS, struct zapi_pw_status *pw)
+int zebra_read_pw_status_update(ZAPI_CALLBACK_ARGS, struct zapi_pw_status *pw)
 {
 	struct stream *s;
 
@@ -2769,8 +2947,12 @@ void zebra_read_pw_status_update(ZAPI_CALLBACK_ARGS, struct zapi_pw_status *pw)
 
 	/* Get data. */
 	stream_get(pw->ifname, s, IF_NAMESIZE);
-	pw->ifindex = stream_getl(s);
-	pw->status = stream_getl(s);
+	STREAM_GETL(s, pw->ifindex);
+	STREAM_GETL(s, pw->status);
+
+	return 0;
+stream_failure:
+	return -1;
 }
 
 static void zclient_capability_decode(ZAPI_CALLBACK_ARGS)
@@ -2781,7 +2963,14 @@ static void zclient_capability_decode(ZAPI_CALLBACK_ARGS)
 	uint8_t mpls_enabled;
 
 	STREAM_GETL(s, vrf_backend);
-	vrf_configure_backend(vrf_backend);
+
+	if (vrf_backend < 0 || vrf_configure_backend(vrf_backend)) {
+		flog_err(EC_LIB_ZAPI_ENCODE,
+			 "%s: Garbage VRF backend type: %d\n", __func__,
+			 vrf_backend);
+		goto stream_failure;
+	}
+
 
 	memset(&cap, 0, sizeof(cap));
 	STREAM_GETC(s, mpls_enabled);