[PATCH net 0/2] tipc: fix deadlock and NUll pointer dereference in tipc_named_node_up()

[PATCH net 0/2] tipc: fix deadlock and NUll pointer dereference in tipc_named_node_up()

[Tung Nguyen]

Patch 1 fixes a self-deadlock when inspecting `tipc_named_node_up()`.

Patch 2 fixes a NULL pointer dereference caused by a race between the
`tipc_net_finalize` work and `tipc_named_node_up()`. This issue was
also reported by syzbot.

Tung Nguyen (2):
  tipc: fix self-deadlock in tipc_named_node_up
  tipc: fix NULL pointer dereference in named_distribute()

 net/tipc/core.c       |   3 +
 net/tipc/core.h       |   6 ++
 net/tipc/name_distr.c | 148 +++++++++++++++++++++++++++----
 net/tipc/name_distr.h |   3 +-
 net/tipc/name_table.c | 135 +++++++++++++++++++++++++---
 net/tipc/name_table.h |   8 +-
 net/tipc/net.c        |   9 +-
 net/tipc/node.c       | 199 +++++++++++++++++++++++++++++++++---------
 net/tipc/node.h       |   3 +
 net/tipc/socket.c     |   2 +-
 10 files changed, 441 insertions(+), 75 deletions(-)

-- 
2.53.0

[PATCH net 1/2] tipc: fix self-deadlock in tipc_named_node_up

[Tung Nguyen]

The following deadlock can occur, as observed during code inspection:

tipc_named_node_up()
 read_lock_bh(&nt->cluster_scope_lock) <-- acquire read lock
 named_distribute()
 tipc_node_xmit()
  tipc_node_link_down()
   tipc_node_write_unlock()
    tipc_nametbl_withdraw()
     spin_lock_bh(&tn->nametbl_lock)
      tipc_named_withdraw()
       write_lock_bh(&nt->cluster_scope_lock) <-- acquire write lock

Fix this issue by:
- Removing cluster_scope_lock, which protects the name table's cluster_scope
  list. Use nametbl_lock instead to protect the list against races when multiple
  threads concurrently call tipc_named_publish(), tipc_named_withdraw(),
  and tipc_named_node_up().

- Introducing a new function, tipc_node_distribute(), which is similar to
  tipc_node_xmit() but does not call tipc_node_link_down(). Both functions call
  the newly introduced tipc_node_build_list(), which builds a list of messages
  in sequence.

Fixes: 988f3f1603d4 ("tipc: eliminate message disordering during binding table update")
Signed-off-by: Tung Nguyen <tung.quang.nguyen@est.tech>
---
 net/tipc/name_distr.c |  32 ++++++----
 net/tipc/name_distr.h |   2 +-
 net/tipc/name_table.c |   1 -
 net/tipc/name_table.h |   2 -
 net/tipc/node.c       | 140 ++++++++++++++++++++++++++++++++----------
 net/tipc/node.h       |   2 +
 6 files changed, 130 insertions(+), 49 deletions(-)

diff --git a/net/tipc/name_distr.c b/net/tipc/name_distr.c
index ba4f4906e13b..a60f213ed118 100644
--- a/net/tipc/name_distr.c
+++ b/net/tipc/name_distr.c
@@ -95,9 +95,7 @@ struct sk_buff *tipc_named_publish(struct net *net, struct publication *p)
 		list_add_tail_rcu(&p->binding_node, &nt->node_scope);
 		return NULL;
 	}
-	write_lock_bh(&nt->cluster_scope_lock);
 	list_add_tail(&p->binding_node, &nt->cluster_scope);
-	write_unlock_bh(&nt->cluster_scope_lock);
 	skb = named_prepare_buf(net, PUBLICATION, ITEM_SIZE, 0);
 	if (!skb) {
 		pr_warn("Publication distribution failure\n");
@@ -121,9 +119,7 @@ struct sk_buff *tipc_named_withdraw(struct net *net, struct publication *p)
 	struct distr_item *item;
 	struct sk_buff *skb;
 
-	write_lock_bh(&nt->cluster_scope_lock);
 	list_del(&p->binding_node);
-	write_unlock_bh(&nt->cluster_scope_lock);
 	if (p->scope == TIPC_NODE_SCOPE)
 		return NULL;
 
@@ -202,25 +198,39 @@ static void named_distribute(struct net *net, struct sk_buff_head *list,
  * @net: the associated network namespace
  * @dnode: destination node
  * @capabilities: peer node's capabilities
+ *
+ * Return:
+ * * 0               - Success
+ * * -ENOBUFS        - No buffer space is available
+ * * -EHOSTUNREACH   - Destination node does not exist or cannot be reached
+ *
  */
-void tipc_named_node_up(struct net *net, u32 dnode, u16 capabilities)
+int tipc_named_node_up(struct net *net, u32 dnode, u16 capabilities)
 {
 	struct name_table *nt = tipc_name_table(net);
 	struct tipc_net *tn = tipc_net(net);
 	struct sk_buff_head head;
-	u16 seqno;
+	int bearer_id = 0;
+	int rc = 0;
 
 	__skb_queue_head_init(&head);
 	spin_lock_bh(&tn->nametbl_lock);
 	if (!(capabilities & TIPC_NAMED_BCAST))
 		nt->rc_dests++;
-	seqno = nt->snd_nxt;
+	named_distribute(net, &head, dnode, &nt->cluster_scope, nt->snd_nxt);
+	if (!skb_queue_empty(&head))
+		rc = tipc_node_distribute(net, &head, dnode, &bearer_id, 0);
 	spin_unlock_bh(&tn->nametbl_lock);
 
-	read_lock_bh(&nt->cluster_scope_lock);
-	named_distribute(net, &head, dnode, &nt->cluster_scope, seqno);
-	tipc_node_xmit(net, &head, dnode, 0);
-	read_unlock_bh(&nt->cluster_scope_lock);
+	/* The link is congested after the list is inserted into
+	 * the link's send queue. Return 0, as this is normal.
+	 */
+	if (rc == -ELINKCONG)
+		rc = 0;
+	else if (rc == -EMSGSIZE)
+		rc = -ENOBUFS;
+
+	return rc;
 }
 
 /**
diff --git a/net/tipc/name_distr.h b/net/tipc/name_distr.h
index c677f6f082df..3f41e36efee3 100644
--- a/net/tipc/name_distr.h
+++ b/net/tipc/name_distr.h
@@ -69,7 +69,7 @@ struct distr_item {
 
 struct sk_buff *tipc_named_publish(struct net *net, struct publication *publ);
 struct sk_buff *tipc_named_withdraw(struct net *net, struct publication *publ);
-void tipc_named_node_up(struct net *net, u32 dnode, u16 capabilities);
+int tipc_named_node_up(struct net *net, u32 dnode, u16 capabilities);
 void tipc_named_rcv(struct net *net, struct sk_buff_head *namedq,
 		    u16 *rcv_nxt, bool *open);
 void tipc_named_reinit(struct net *net);
diff --git a/net/tipc/name_table.c b/net/tipc/name_table.c
index 6fda36ab1766..8924bdb6a443 100644
--- a/net/tipc/name_table.c
+++ b/net/tipc/name_table.c
@@ -921,7 +921,6 @@ int tipc_nametbl_init(struct net *net)
 
 	INIT_LIST_HEAD(&nt->node_scope);
 	INIT_LIST_HEAD(&nt->cluster_scope);
-	rwlock_init(&nt->cluster_scope_lock);
 	tn->nametbl = nt;
 	spin_lock_init(&tn->nametbl_lock);
 	return 0;
diff --git a/net/tipc/name_table.h b/net/tipc/name_table.h
index 7ff6eeebaae6..afe665e68bb1 100644
--- a/net/tipc/name_table.h
+++ b/net/tipc/name_table.h
@@ -97,7 +97,6 @@ struct publication {
  * @cluster_scope: all local publications with cluster scope
  *               - used by name_distr to send bulk updates to new nodes
  *               - used by name_distr during re-init of name table
- * @cluster_scope_lock: lock for accessing @cluster_scope
  * @local_publ_count: number of publications issued by this node
  * @rc_dests: destination node counter
  * @snd_nxt: next sequence number to be used
@@ -107,7 +106,6 @@ struct name_table {
 	struct hlist_head services[TIPC_NAMETBL_SIZE];
 	struct list_head node_scope;
 	struct list_head cluster_scope;
-	rwlock_t cluster_scope_lock;
 	u32 local_publ_count;
 	u32 rc_dests;
 	u32 snd_nxt;
diff --git a/net/tipc/node.c b/net/tipc/node.c
index 53b0993c2c23..b19d6652dc73 100644
--- a/net/tipc/node.c
+++ b/net/tipc/node.c
@@ -402,6 +402,7 @@ static void tipc_node_write_unlock(struct tipc_node *n)
 	struct list_head *publ_list;
 	struct tipc_uaddr ua;
 	u32 bearer_id, node;
+	int rc = 0;
 
 	if (likely(!flags)) {
 		write_unlock_bh(&n->lock);
@@ -425,7 +426,7 @@ static void tipc_node_write_unlock(struct tipc_node *n)
 		tipc_publ_notify(net, publ_list, node, n->capabilities);
 
 	if (flags & TIPC_NOTIFY_NODE_UP)
-		tipc_named_node_up(net, node, n->capabilities);
+		rc = tipc_named_node_up(net, node, n->capabilities);
 
 	if (flags & TIPC_NOTIFY_LINK_UP) {
 		tipc_mon_peer_up(net, node, bearer_id);
@@ -435,6 +436,9 @@ static void tipc_node_write_unlock(struct tipc_node *n)
 		tipc_mon_peer_down(net, node, bearer_id);
 		tipc_nametbl_withdraw(net, &ua, &sk, sk.ref);
 	}
+
+	if (rc == -ENOBUFS)
+		tipc_node_link_down(n, bearer_id, false);
 }
 
 static void tipc_node_assign_peer_net(struct tipc_node *n, u32 hash_mixes)
@@ -1682,6 +1686,51 @@ static void tipc_lxc_xmit(struct net *peer_net, struct sk_buff_head *list)
 	}
 }
 
+static int tipc_node_build_list(struct tipc_node *n, struct sk_buff_head *list,
+				struct sk_buff_head *xmitq, int selector,
+				struct tipc_media_addr *maddr,
+				int *bearer_id, bool *lxc)
+{
+	struct tipc_link_entry *le = NULL;
+	bool node_up = false;
+	struct net *peer_net;
+	int rc = 0;
+
+	rcu_read_lock();
+	tipc_node_read_lock(n);
+	node_up = node_is_up(n);
+	peer_net = n->peer_net;
+	tipc_node_read_unlock(n);
+
+	if (node_up && peer_net && check_net(peer_net)) {
+		/* xmit inner linux container */
+		tipc_lxc_xmit(peer_net, list);
+		if (likely(skb_queue_empty(list))) {
+			rcu_read_unlock();
+			*lxc = true;
+			return 0;
+		}
+	}
+	rcu_read_unlock();
+
+	tipc_node_read_lock(n);
+	*bearer_id = n->active_links[selector & 1];
+	if (unlikely(*bearer_id == INVALID_BEARER_ID)) {
+		tipc_node_read_unlock(n);
+		__skb_queue_purge(list);
+		return -EHOSTUNREACH;
+	}
+
+	le = &n->links[*bearer_id];
+	*maddr = le->maddr;
+	spin_lock_bh(&le->lock);
+	rc = tipc_link_xmit(le->link, list, xmitq);
+	spin_unlock_bh(&le->lock);
+	tipc_node_read_unlock(n);
+
+	return rc;
+}
+
 /**
  * tipc_node_xmit() - general link level function for message sending
  * @net: the applicable net namespace
@@ -1694,12 +1743,11 @@ static void tipc_lxc_xmit(struct net *peer_net, struct sk_buff_head *list)
 int tipc_node_xmit(struct net *net, struct sk_buff_head *list,
 		   u32 dnode, int selector)
 {
-	struct tipc_link_entry *le = NULL;
-	struct tipc_node *n;
+	struct tipc_media_addr maddr;
 	struct sk_buff_head xmitq;
-	bool node_up = false;
-	struct net *peer_net;
-	int bearer_id;
+	struct tipc_node *n;
+	int bearer_id = 0;
+	bool lxc = false;
 	int rc;
 
 	if (in_own_node(net, dnode)) {
@@ -1715,42 +1763,66 @@ int tipc_node_xmit(struct net *net, struct sk_buff_head *list,
 		return -EHOSTUNREACH;
 	}
 
-	rcu_read_lock();
-	tipc_node_read_lock(n);
-	node_up = node_is_up(n);
-	peer_net = n->peer_net;
-	tipc_node_read_unlock(n);
-	if (node_up && peer_net && check_net(peer_net)) {
-		/* xmit inner linux container */
-		tipc_lxc_xmit(peer_net, list);
-		if (likely(skb_queue_empty(list))) {
-			rcu_read_unlock();
-			tipc_node_put(n);
-			return 0;
-		}
-	}
-	rcu_read_unlock();
+	__skb_queue_head_init(&xmitq);
+	rc = tipc_node_build_list(n, list, &xmitq, selector, &maddr,
+				  &bearer_id, &lxc);
 
-	tipc_node_read_lock(n);
-	bearer_id = n->active_links[selector & 1];
-	if (unlikely(bearer_id == INVALID_BEARER_ID)) {
-		tipc_node_read_unlock(n);
+	if (lxc) {
 		tipc_node_put(n);
+		return 0;
+	}
+
+	if (unlikely(rc == -ENOBUFS))
+		tipc_node_link_down(n, bearer_id, false);
+	else
+		tipc_bearer_xmit(net, bearer_id, &xmitq, &maddr, n);
+
+	tipc_node_put(n);
+
+	return rc;
+}
+
+/**
+ * tipc_node_distribute() - distribute list of messages to specific destination
+ * @net: the applicable net namespace
+ * @list: chain of buffers containing message
+ * @dnode: address of destination node
+ * @bearer_id: id of the bearer
+ * @selector: a number used for deterministic link selection
+ *
+ * Return:
+ * * 0              - Success
+ * * -ELINKCONG     - Link is congested
+ * * -EHOSTUNREACH  - Host is not reachable
+ * * -EMSGSIZE      - Message's size is too big
+ * * -ENOBUF        - No buffer space is available
+ *
+ */
+int tipc_node_distribute(struct net *net, struct sk_buff_head *list,
+			 u32 dnode, int *bearer_id, int selector)
+{
+	struct tipc_media_addr maddr;
+	struct sk_buff_head xmitq;
+	struct tipc_node *n;
+	bool lxc = false;
+	int rc;
+
+	n = tipc_node_find(net, dnode);
+	if (unlikely(!n)) {
 		__skb_queue_purge(list);
 		return -EHOSTUNREACH;
 	}
 
 	__skb_queue_head_init(&xmitq);
-	le = &n->links[bearer_id];
-	spin_lock_bh(&le->lock);
-	rc = tipc_link_xmit(le->link, list, &xmitq);
-	spin_unlock_bh(&le->lock);
-	tipc_node_read_unlock(n);
+	rc = tipc_node_build_list(n, list, &xmitq, selector, &maddr,
+				  bearer_id, &lxc);
+	if (lxc) {
+		tipc_node_put(n);
+		return 0;
+	}
 
-	if (unlikely(rc == -ENOBUFS))
-		tipc_node_link_down(n, bearer_id, false);
-	else
-		tipc_bearer_xmit(net, bearer_id, &xmitq, &le->maddr, n);
+	if (likely(rc != -ENOBUFS))
+		tipc_bearer_xmit(net, *bearer_id, &xmitq, &maddr, n);
 
 	tipc_node_put(n);
 
diff --git a/net/tipc/node.h b/net/tipc/node.h
index 154a5bbb0d29..775a8ec240fb 100644
--- a/net/tipc/node.h
+++ b/net/tipc/node.h
@@ -100,6 +100,8 @@ int tipc_node_get_linkname(struct net *net, u32 bearer_id, u32 node,
 			   char *linkname, size_t len);
 int tipc_node_xmit(struct net *net, struct sk_buff_head *list, u32 dnode,
 		   int selector);
+int tipc_node_distribute(struct net *net, struct sk_buff_head *list,
+			 u32 dnode, int *bearer_id, int selector);
 int tipc_node_distr_xmit(struct net *net, struct sk_buff_head *list);
 int tipc_node_xmit_skb(struct net *net, struct sk_buff *skb, u32 dest,
 		       u32 selector);
-- 
2.53.0

[PATCH net 2/2] tipc: fix NULL pointer dereference in named_distribute()

[Tung Nguyen]

Syzbot reports NULL pointer dereference in named_distribute():

...
RIP: 0010:buf_msg net/tipc/msg.h:202 [inline]
RIP: 0010:named_distribute net/tipc/name_distr.c:195 [inline]
RIP: 0010:tipc_named_node_up+0x8e0/0xb00 net/tipc/name_distr.c:221
...
tipc_node_write_unlock+0x2ce/0x4f0 net/tipc/node.c:428
tipc_rcv+0xb55/0x1020 net/tipc/node.c:2196
tipc_l2_rcv_msg+0x2d9/0x470 net/tipc/bearer.c:671

The root cause is racing between tipc_net_finalize() and
named_distribute() as follows:

   Thread 1 (tipc_net_finalize) | Thread 2 (named_distribute)
   -----------------------------|-----------------------------
                                | ...
                                | list_for_each_entry(publ, pls, binding_node) {
                                | ...
                                | __skb_queue_tail(list, skb);
                                | ...
                                | }
                                | ...
                                | hdr = buf_msg(skb_peek_tail(list));
   ...                          |
   tipc_nametbl_publish();      |

If 'tipc_nametbl_publish()' (Thread 1) executes after 'named_distribute()'
, list will be empty. As a result, NULL is passed to 'buf_msg()', leading
to a NULL pointer dereference.

Fix the this issue by checking for the node's self-address publication on
either node-up or node-timeout event. If the publication does not exist, insert
it into the name table and publish it to the other nodes.

To facilitate the implementation, three major changes are made:
- Move the sc->lock acquisition out of tipc_service_insert_publ() and into its
  callers.
- Add an output parameter (*err) to tipc_nametbl_publish() to detect memory
  allocation errors.
- Add three fields (nt_stop, nt_self_node_exist, and node_addr_set) to struct
  tipc_net to synchronize the insertion and deletion of the node's self-address
  publication.

Reported-by: syzbot+64197f7577a09686ffdb@syzkaller.appspotmail.com
Closes: https://syzkaller.appspot.com/bug?extid=64197f7577a09686ffdb
Fixes: adba75be0d23 ("tipc: fix lockdep warning when reinitilaizing sockets")
Signed-off-by: Tung Nguyen <tung.quang.nguyen@est.tech>
---
 net/tipc/core.c       |   3 +
 net/tipc/core.h       |   6 ++
 net/tipc/name_distr.c | 120 ++++++++++++++++++++++++++++++++++---
 net/tipc/name_distr.h |   1 +
 net/tipc/name_table.c | 134 ++++++++++++++++++++++++++++++++++++++----
 net/tipc/name_table.h |   6 +-
 net/tipc/net.c        |   9 ++-
 net/tipc/node.c       |  61 ++++++++++++++++---
 net/tipc/node.h       |   1 +
 net/tipc/socket.c     |   2 +-
 10 files changed, 314 insertions(+), 29 deletions(-)

diff --git a/net/tipc/core.c b/net/tipc/core.c
index 315975c3be81..0f1d3ddeae09 100644
--- a/net/tipc/core.c
+++ b/net/tipc/core.c
@@ -61,6 +61,9 @@ static int __net_init tipc_init_net(struct net *net)
 	tn->trial_addr = 0;
 	tn->addr_trial_end = 0;
 	tn->capabilities = TIPC_NODE_CAPABILITIES;
+	tn->nt_stop = false;
+	tn->nt_self_node_exist = false;
+	atomic_set(&tn->node_addr_set, 0);
 	INIT_WORK(&tn->work, tipc_net_finalize_work);
 	memset(tn->node_id, 0, sizeof(tn->node_id));
 	memset(tn->node_id_string, 0, sizeof(tn->node_id_string));
diff --git a/net/tipc/core.h b/net/tipc/core.h
index 9ce5f9ff6cc0..975b1d48186e 100644
--- a/net/tipc/core.h
+++ b/net/tipc/core.h
@@ -145,6 +145,12 @@ struct tipc_net {
 	struct work_struct work;
 	/* The numbers of work queues in schedule */
 	atomic_t wq_count;
+	/* Flag to indicate that the node's address is set */
+	atomic_t node_addr_set;
+	/* The name table has stopped working */
+	bool nt_stop;
+	/* Node's self address publication is available in the name table */
+	bool nt_self_node_exist;
 };
 
 static inline struct tipc_net *tipc_net(struct net *net)
diff --git a/net/tipc/name_distr.c b/net/tipc/name_distr.c
index a60f213ed118..345bd767983e 100644
--- a/net/tipc/name_distr.c
+++ b/net/tipc/name_distr.c
@@ -142,9 +142,14 @@ struct sk_buff *tipc_named_withdraw(struct net *net, struct publication *p)
  * @dnode: node to be updated
  * @pls: linked list of publication items to be packed into buffer chain
  * @seqno: sequence number for this message
+ *
+ * Return:
+ * * 0          - Success
+ * * -ENOBUFS   - No buffer space is available
+ *
  */
-static void named_distribute(struct net *net, struct sk_buff_head *list,
-			     u32 dnode, struct list_head *pls, u16 seqno)
+static int named_distribute(struct net *net, struct sk_buff_head *list,
+			    u32 dnode, struct list_head *pls, u16 seqno)
 {
 	struct publication *publ;
 	struct sk_buff *skb = NULL;
@@ -160,8 +165,9 @@ static void named_distribute(struct net *net, struct sk_buff_head *list,
 			skb = named_prepare_buf(net, PUBLICATION, msg_rem,
 						dnode);
 			if (!skb) {
+				__skb_queue_purge(list);
 				pr_warn("Bulk publication failure\n");
-				return;
+				return -ENOBUFS;
 			}
 			hdr = buf_msg(skb);
 			msg_set_bc_ack_invalid(hdr, true);
@@ -191,6 +197,94 @@ static void named_distribute(struct net *net, struct sk_buff_head *list,
 	hdr = buf_msg(skb_peek_tail(list));
 	msg_set_last_bulk(hdr);
 	msg_set_named_seqno(hdr, seqno);
+
+	return 0;
+}
+
+/**
+ * __tipc_named_distribute - build a list of publications need to be distributed
+ * @net: the associated network namespace
+ * @dnode: destination node
+ * @head: list of publications
+ *
+ * Return:
+ * * 0          - Success
+ * * -ENOBUFS   - No buffer space is available
+ *
+ */
+static int __tipc_named_distribute(struct net *net, u32 dnode,
+				   struct sk_buff_head *head)
+{
+	struct name_table *nt = tipc_name_table(net);
+	struct tipc_net *tn = tipc_net(net);
+
+	/* Name table has been deleted after namespace teardown or
+	 * TIPC module exit.
+	 */
+	if (unlikely(tn->nt_stop))
+		return 0;
+
+	/* Node's self-address is not set yet */
+	if (!atomic_read(&tn->node_addr_set)) {
+		/* Defer the distribution to node's timer */
+		tipc_node_set_dist_bulk(net, dnode);
+		return 0;
+	}
+
+	/* Previous call to tipc_nametb_insert_self_node_pub() was successful.
+	 * The node's self-address publication was added to 'nt->cluster_scope'.
+	 * So, 'nt->cluster_scope' is not empty now. Go to distribution.
+	 */
+	if (tn->nt_self_node_exist)
+		goto distribute;
+
+	if (tipc_nametb_insert_self_node_pub(net))
+		return -ENOBUFS;
+
+	tn->nt_self_node_exist = true;
+
+distribute:
+	if (named_distribute(net, head, dnode,
+			     &nt->cluster_scope, nt->snd_nxt) == -ENOBUFS)
+		return -ENOBUFS;
+
+	return 0;
+}
+
+/**
+ * tipc_named_distribute - distribute all publications to specified node
+ * @net: the associated network namespace
+ * @dnode: destination node
+ *
+ * Return:
+ * * 0               - Success
+ * * -ENOBUFS        - No buffer space is available
+ * * -EHOSTUNREACH   - Destination node does not exist or cannot be reached
+ *
+ */
+int tipc_named_distribute(struct net *net, u32 dnode)
+{
+	struct tipc_net *tn = tipc_net(net);
+	struct sk_buff_head head;
+	int bearer_id = 0;
+	int rc = 0;
+
+	__skb_queue_head_init(&head);
+	spin_lock_bh(&tn->nametbl_lock);
+	rc = __tipc_named_distribute(net, dnode, &head);
+	if (!rc && !skb_queue_empty(&head))
+		rc = tipc_node_distribute(net, &head, dnode, &bearer_id, 0);
+	spin_unlock_bh(&tn->nametbl_lock);
+
+	/* The link is congested after the list is inserted into
+	 * the link's send queue. Return 0, as this is normal.
+	 */
+	if (rc == -ELINKCONG)
+		rc = 0;
+	else if (rc == -EMSGSIZE)
+		rc = -ENOBUFS;
+
+	return rc;
 }
 
 /**
@@ -217,8 +311,8 @@ int tipc_named_node_up(struct net *net, u32 dnode, u16 capabilities)
 	spin_lock_bh(&tn->nametbl_lock);
 	if (!(capabilities & TIPC_NAMED_BCAST))
 		nt->rc_dests++;
-	named_distribute(net, &head, dnode, &nt->cluster_scope, nt->snd_nxt);
-	if (!skb_queue_empty(&head))
+	rc = __tipc_named_distribute(net, dnode, &head);
+	if (!rc && !skb_queue_empty(&head))
 		rc = tipc_node_distribute(net, &head, dnode, &bearer_id, 0);
 	spin_unlock_bh(&tn->nametbl_lock);
 
@@ -289,10 +383,12 @@ void tipc_publ_notify(struct net *net, struct list_head *nsub_list,
 static bool tipc_update_nametbl(struct net *net, struct distr_item *i,
 				u32 node, u32 dtype)
 {
+	struct tipc_net *tn = tipc_net(net);
 	struct publication *p = NULL;
 	u32 lower = ntohl(i->lower);
 	u32 upper = ntohl(i->upper);
 	struct tipc_socket_addr sk;
+	u32 type = ntohl(i->type);
 	u32 key = ntohl(i->key);
 	struct tipc_uaddr ua;
 
@@ -303,13 +399,22 @@ static bool tipc_update_nametbl(struct net *net, struct distr_item *i,
 	if (lower > upper)
 		return false;
 
+	/* Not insert or remove duplicate node-up event publication sent from
+	 * remote nodes
+	 */
+	if (unlikely(type == TIPC_NODE_STATE && tn->node_addr &&
+		     lower == tn->node_addr && upper == tn->node_addr)) {
+		pr_warn_ratelimited("Can't update duplicate node-up pub\n");
+		return false;
+	}
+
 	tipc_uaddr(&ua, TIPC_SERVICE_RANGE, TIPC_CLUSTER_SCOPE,
-		   ntohl(i->type), lower, upper);
+		   type, lower, upper);
 	sk.ref = ntohl(i->port);
 	sk.node = node;
 
 	if (dtype == PUBLICATION) {
-		p = tipc_nametbl_insert_publ(net, &ua, &sk, key);
+		p = tipc_nametbl_insert_publ(net, &ua, &sk, key, NULL);
 		if (p) {
 			tipc_node_subscribe(net, &p->binding_node, node);
 			return true;
@@ -426,5 +531,6 @@ void tipc_named_reinit(struct net *net)
 	list_for_each_entry_rcu(p, &nt->cluster_scope, binding_node)
 		p->sk.node = self;
 	nt->rc_dests = 0;
+	tn->nt_stop = false;
 	spin_unlock_bh(&tn->nametbl_lock);
 }
diff --git a/net/tipc/name_distr.h b/net/tipc/name_distr.h
index 3f41e36efee3..2043096ae967 100644
--- a/net/tipc/name_distr.h
+++ b/net/tipc/name_distr.h
@@ -75,5 +75,6 @@ void tipc_named_rcv(struct net *net, struct sk_buff_head *namedq,
 void tipc_named_reinit(struct net *net);
 void tipc_publ_notify(struct net *net, struct list_head *nsub_list,
 		      u32 addr, u16 capabilities);
+int tipc_named_distribute(struct net *net, u32 dnode);
 
 #endif
diff --git a/net/tipc/name_table.c b/net/tipc/name_table.c
index 8924bdb6a443..3bcb8762bc9c 100644
--- a/net/tipc/name_table.c
+++ b/net/tipc/name_table.c
@@ -329,7 +329,7 @@ static struct service_range *tipc_service_create_range(struct tipc_service *sc,
 
 static bool tipc_service_insert_publ(struct net *net,
 				     struct tipc_service *sc,
-				     struct publication *p)
+				     struct publication *p, int *err)
 {
 	struct tipc_subscription *sub, *tmp;
 	struct service_range *sr;
@@ -339,10 +339,12 @@ static bool tipc_service_insert_publ(struct net *net,
 	bool res = false;
 	u32 key = p->key;
 
-	spin_lock_bh(&sc->lock);
 	sr = tipc_service_create_range(sc, p);
-	if (!sr)
+	if (!sr) {
+		if (err)
+			*err = -ENOBUFS;
 		goto  exit;
+	}
 
 	first = list_empty(&sr->all_publ);
 
@@ -353,6 +355,8 @@ static bool tipc_service_insert_publ(struct net *net,
 			pr_debug("Failed to bind duplicate %u,%u,%u/%u:%u/%u\n",
 				 p->sr.type, p->sr.lower, p->sr.upper,
 				 node, p->sk.ref, key);
+			if (err)
+				*err = -EEXIST;
 			goto exit;
 		}
 	}
@@ -371,7 +375,6 @@ static bool tipc_service_insert_publ(struct net *net,
 	if (!res)
 		pr_warn("Failed to bind to %u,%u,%u\n",
 			p->sr.type, p->sr.lower, p->sr.upper);
-	spin_unlock_bh(&sc->lock);
 	return res;
 }
 
@@ -478,20 +481,33 @@ static struct tipc_service *tipc_service_find(struct net *net,
 struct publication *tipc_nametbl_insert_publ(struct net *net,
 					     struct tipc_uaddr *ua,
 					     struct tipc_socket_addr *sk,
-					     u32 key)
+					     u32 key, int *err)
 {
 	struct tipc_service *sc;
 	struct publication *p;
 
 	p = tipc_publ_create(ua, sk, key);
-	if (!p)
+	if (!p) {
+		if (err)
+			*err = -ENOBUFS;
 		return NULL;
+	}
 
 	sc = tipc_service_find(net, ua);
 	if (!sc)
 		sc = tipc_service_create(net, ua);
-	if (sc && tipc_service_insert_publ(net, sc, p))
-		return p;
+
+	if (sc) {
+		spin_lock_bh(&sc->lock);
+		if (tipc_service_insert_publ(net, sc, p, err)) {
+			spin_unlock_bh(&sc->lock);
+			return p;
+		}
+		spin_unlock_bh(&sc->lock);
+	} else if (err) {
+		*err = -ENOBUFS;
+	}
+
 	kfree(p);
 	return NULL;
 }
@@ -760,7 +776,8 @@ void tipc_nametbl_build_group(struct net *net, struct tipc_group *grp,
 /* tipc_nametbl_publish - add service binding to name table
  */
 struct publication *tipc_nametbl_publish(struct net *net, struct tipc_uaddr *ua,
-					 struct tipc_socket_addr *sk, u32 key)
+					 struct tipc_socket_addr *sk,
+					 u32 key, int *err)
 {
 	struct name_table *nt = tipc_name_table(net);
 	u32 max_user_pub = TIPC_MAX_PUBL - 1;
@@ -768,6 +785,7 @@ struct publication *tipc_nametbl_publish(struct net *net, struct tipc_uaddr *ua,
 	struct publication *p = NULL;
 	struct sk_buff *skb = NULL;
 	bool protocol_type = false;
+	int error = 0;
 	u32 rc_dests;
 
 	if (ua->sr.type == TIPC_NODE_STATE || ua->sr.type == TIPC_LINK_STATE ||
@@ -788,7 +806,7 @@ struct publication *tipc_nametbl_publish(struct net *net, struct tipc_uaddr *ua,
 	}
 
 insert:
-	p = tipc_nametbl_insert_publ(net, ua, sk, key);
+	p = tipc_nametbl_insert_publ(net, ua, sk, key, &error);
 	if (p) {
 		/* Not count node state, link state and topology server types
 		 * so that maximum nt->local_publ_count does not prevent
@@ -798,6 +816,13 @@ struct publication *tipc_nametbl_publish(struct net *net, struct tipc_uaddr *ua,
 		if (!protocol_type)
 			nt->local_publ_count++;
 		skb = tipc_named_publish(net, p);
+		/* Local-scope publication is not published (skb is NULL), only
+		 * cluster-cope one is.
+		 */
+		if (!skb && p->scope == TIPC_CLUSTER_SCOPE && err)
+			*err = -ENOBUFS;
+	} else if ((error == -ENOBUFS) && err) {
+		*err = -ENOBUFS;
 	}
 	rc_dests = nt->rc_dests;
 exit:
@@ -809,6 +834,83 @@ struct publication *tipc_nametbl_publish(struct net *net, struct tipc_uaddr *ua,
 
 }
 
+/**
+ * tipc_nametb_insert_self_node_pub - insert publication of node's self address
+ * @net: network namespace
+ *
+ * Return:
+ * * 0          - Success
+ * * -ENOBUFS   - No buffer space is available
+ *
+ */
+int tipc_nametb_insert_self_node_pub(struct net *net)
+{
+	struct name_table *nt = tipc_name_table(net);
+	struct tipc_net *tn = tipc_net(net);
+	struct tipc_socket_addr sk;
+	struct service_range *sr;
+	struct tipc_service *sc;
+	bool sc_created = false;
+	struct publication *p;
+	struct tipc_uaddr ua;
+	int err = 0;
+
+	sk.ref = 0;
+	sk.node = tn->node_addr;
+	tipc_uaddr(&ua, TIPC_SERVICE_RANGE, TIPC_CLUSTER_SCOPE,
+		   TIPC_NODE_STATE, tn->node_addr, tn->node_addr);
+	sc = tipc_service_find(net, &ua);
+	if (!sc) {
+		sc = tipc_service_create(net, &ua);
+		if (!sc)
+			return -ENOBUFS;
+		sc_created = true;
+	}
+
+	spin_lock_bh(&sc->lock);
+	sr = tipc_service_find_range(sc, &ua);
+	/* 1. tipc_net_finalize_work() is not scheduled because of namespace
+	 *    teardown.
+	 * 2. Or tipc_net_finalize() ---> tipc_nametbl_publish() has failed
+	 *    to insert node self address publication into nt->cluster_scope
+	 *    due to memory allocation failure.
+	 * 3. Or tipc_net_finalize() ---> tipc_nametbl_publish() has not
+	 *    executed yet.
+	 */
+	if (!sr) {
+		p = tipc_publ_create(&ua, &sk, tn->node_addr);
+		if (!p)
+			goto error;
+
+		if (!tipc_service_insert_publ(net, sc, p, &err)) {
+			kfree(p);
+			goto error;
+		}
+
+		list_add_tail(&p->binding_node, &nt->cluster_scope);
+		goto exit;
+error:
+		if (sc_created) {
+			hlist_del_init_rcu(&sc->service_list);
+			spin_unlock_bh(&sc->lock);
+			kfree_rcu(sc, rcu);
+			if (unlikely(err == -EEXIST))
+				return 0;
+			else
+				return -ENOBUFS;
+		}
+
+		spin_unlock_bh(&sc->lock);
+		if (unlikely(err == -EEXIST))
+			return 0;
+		else
+			return -ENOBUFS;
+	}
+exit:
+	spin_unlock_bh(&sc->lock);
+	return 0;
+}
+
 /**
  * tipc_nametbl_withdraw - withdraw a service binding
  * @net: network namespace
@@ -933,6 +1035,7 @@ int tipc_nametbl_init(struct net *net)
  */
 static void tipc_service_delete(struct net *net, struct tipc_service *sc)
 {
+	struct tipc_net *tn = tipc_net(net);
 	struct service_range *sr, *tmpr;
 	struct publication *p, *tmp;
 
@@ -940,6 +1043,15 @@ static void tipc_service_delete(struct net *net, struct tipc_service *sc)
 	rbtree_postorder_for_each_entry_safe(sr, tmpr, &sc->ranges, tree_node) {
 		list_for_each_entry_safe(p, tmp, &sr->all_publ, all_publ) {
 			tipc_service_remove_publ(sr, &p->sk, p->key);
+			/* tipc_nametbl_withdraw() does not delete
+			 * 'p->binding_node' with type TIPC_NODE_STATE and
+			 * range [tn->node_addr, tn->node_addr] from
+			 * 'tn->cluster_scope'. So, delete it here.
+			 */
+			if (p->sr.type == TIPC_NODE_STATE &&
+			    p->sr.lower == tn->node_addr &&
+			    p->sr.upper == tn->node_addr)
+				list_del(&p->binding_node);
 			kfree_rcu(p, rcu);
 		}
 		rb_erase_augmented(&sr->tree_node, &sc->ranges, &sr_callbacks);
@@ -970,6 +1082,8 @@ void tipc_nametbl_stop(struct net *net)
 			tipc_service_delete(net, service);
 		}
 	}
+	tn->nt_stop = true;
+	tn->nt_self_node_exist = false;
 	spin_unlock_bh(&tn->nametbl_lock);
 
 	/* TODO: clear tn->nametbl, implement proper RCU rules ? */
diff --git a/net/tipc/name_table.h b/net/tipc/name_table.h
index afe665e68bb1..4b411b6e9e40 100644
--- a/net/tipc/name_table.h
+++ b/net/tipc/name_table.h
@@ -124,13 +124,15 @@ bool tipc_nametbl_lookup_group(struct net *net, struct tipc_uaddr *ua,
 void tipc_nametbl_build_group(struct net *net, struct tipc_group *grp,
 			      struct tipc_uaddr *ua);
 struct publication *tipc_nametbl_publish(struct net *net, struct tipc_uaddr *ua,
-					 struct tipc_socket_addr *sk, u32 key);
+					 struct tipc_socket_addr *sk,
+					 u32 key, int *err);
 void tipc_nametbl_withdraw(struct net *net, struct tipc_uaddr *ua,
 			   struct tipc_socket_addr *sk, u32 key);
 struct publication *tipc_nametbl_insert_publ(struct net *net,
 					     struct tipc_uaddr *ua,
 					     struct tipc_socket_addr *sk,
-					     u32 key);
+					     u32 key, int *err);
+int tipc_nametb_insert_self_node_pub(struct net *net);
 struct publication *tipc_nametbl_remove_publ(struct net *net,
 					     struct tipc_uaddr *ua,
 					     struct tipc_socket_addr *sk,
diff --git a/net/tipc/net.c b/net/tipc/net.c
index 7e65d0b0c4a8..6e8bf0c5ebde 100644
--- a/net/tipc/net.c
+++ b/net/tipc/net.c
@@ -110,11 +110,14 @@ static void tipc_net_finalize(struct net *net, u32 addr);
 
 int tipc_net_init(struct net *net, u8 *node_id, u32 addr)
 {
+	struct tipc_net *tn = tipc_net(net);
+
 	if (tipc_own_id(net)) {
 		pr_info("Cannot configure node identity twice\n");
 		return -1;
 	}
 	pr_info("Started in network mode\n");
+	atomic_set(&tn->node_addr_set, 0);
 
 	if (node_id)
 		tipc_set_node_id(net, node_id);
@@ -138,7 +141,11 @@ static void tipc_net_finalize(struct net *net, u32 addr)
 	tipc_named_reinit(net);
 	tipc_sk_reinit(net);
 	tipc_mon_reinit_self(net);
-	tipc_nametbl_publish(net, &ua, &sk, addr);
+	atomic_set(&tn->node_addr_set, 1);
+	/* Redistribute the node state publication via node-up event
+	 * or node's timer if tipc_nametbl_publish() fails.
+	 */
+	tipc_nametbl_publish(net, &ua, &sk, addr, NULL);
 }
 
 void tipc_net_finalize_work(struct work_struct *work)
diff --git a/net/tipc/node.c b/net/tipc/node.c
index b19d6652dc73..f566472a411f 100644
--- a/net/tipc/node.c
+++ b/net/tipc/node.c
@@ -111,6 +111,7 @@ struct tipc_bclink_entry {
  * @peer_net: peer's net namespace
  * @peer_hash_mix: hash for this peer (FIXME)
  * @crypto_rx: RX crypto handler
+ * @dist_bulk: flag to redistribute cluster-scope publications
  */
 struct tipc_node {
 	u32 addr;
@@ -145,6 +146,7 @@ struct tipc_node {
 #ifdef CONFIG_TIPC_CRYPTO
 	struct tipc_crypto *crypto_rx;
 #endif
+	bool dist_bulk;
 };
 
 /* Node FSM states and events:
@@ -393,6 +395,14 @@ static void tipc_node_write_unlock_fast(struct tipc_node *n)
 	write_unlock_bh(&n->lock);
 }
 
+static void tipc_node_down(struct tipc_node *n)
+{
+	int bearer_id;
+
+	for (bearer_id = 0; bearer_id < MAX_BEARERS; bearer_id++)
+		tipc_node_link_down(n, bearer_id, false);
+}
+
 static void tipc_node_write_unlock(struct tipc_node *n)
 	__releases(n->lock)
 {
@@ -402,7 +412,7 @@ static void tipc_node_write_unlock(struct tipc_node *n)
 	struct list_head *publ_list;
 	struct tipc_uaddr ua;
 	u32 bearer_id, node;
-	int rc = 0;
+	int rc = 0, err = 0;
 
 	if (likely(!flags)) {
 		write_unlock_bh(&n->lock);
@@ -420,6 +430,8 @@ static void tipc_node_write_unlock(struct tipc_node *n)
 	n->action_flags &= ~(TIPC_NOTIFY_NODE_DOWN | TIPC_NOTIFY_NODE_UP |
 			     TIPC_NOTIFY_LINK_DOWN | TIPC_NOTIFY_LINK_UP);
 
+	if (flags & TIPC_NOTIFY_NODE_DOWN)
+		n->dist_bulk = false;
 	write_unlock_bh(&n->lock);
 
 	if (flags & TIPC_NOTIFY_NODE_DOWN)
@@ -430,17 +442,35 @@ static void tipc_node_write_unlock(struct tipc_node *n)
 
 	if (flags & TIPC_NOTIFY_LINK_UP) {
 		tipc_mon_peer_up(net, node, bearer_id);
-		tipc_nametbl_publish(net, &ua, &sk, sk.ref);
+		tipc_nametbl_publish(net, &ua, &sk, sk.ref, &err);
 	}
 	if (flags & TIPC_NOTIFY_LINK_DOWN) {
 		tipc_mon_peer_down(net, node, bearer_id);
 		tipc_nametbl_withdraw(net, &ua, &sk, sk.ref);
 	}
 
+	/* Memory allocation has failed. Bring the node down to start over bulk
+	 * distribution when the first link is up again.
+	 */
 	if (rc == -ENOBUFS)
+		tipc_node_down(n);
+	else if (err == -ENOBUFS)
 		tipc_node_link_down(n, bearer_id, false);
 }
 
+void tipc_node_set_dist_bulk(struct net *net, u32 addr)
+{
+	struct tipc_node *node;
+
+	node = tipc_node_find(net, addr);
+	if (node) {
+		tipc_node_write_lock(node);
+		node->dist_bulk = true;
+		tipc_node_write_unlock_fast(node);
+		tipc_node_put(node);
+	}
+}
+
 static void tipc_node_assign_peer_net(struct tipc_node *n, u32 hash_mixes)
 {
 	int net_id = tipc_netid(n->net);
@@ -568,6 +598,7 @@ struct tipc_node *tipc_node_create(struct net *net, u32 addr, u8 *peer_id,
 	INIT_LIST_HEAD(&n->list);
 	INIT_LIST_HEAD(&n->publ_list);
 	INIT_LIST_HEAD(&n->conn_sks);
+	n->dist_bulk = false;
 	skb_queue_head_init(&n->bc_entry.namedq);
 	skb_queue_head_init(&n->bc_entry.inputq1);
 	__skb_queue_head_init(&n->bc_entry.arrvq);
@@ -807,9 +838,11 @@ static void tipc_node_timeout(struct timer_list *t)
 	struct tipc_node *n = timer_container_of(n, t, timer);
 	struct tipc_link_entry *le;
 	struct sk_buff_head xmitq;
-	int remains = n->link_cnt;
+	bool dist_bulk = false;
+	int remains = 0;
+	int dist_rc = 0;
 	int bearer_id;
-	int rc = 0;
+	int rc;
 
 	trace_tipc_node_timeout(n, false, " ");
 	if (!node_is_up(n) && tipc_node_cleanup(n)) {
@@ -824,13 +857,23 @@ static void tipc_node_timeout(struct timer_list *t)
 #endif
 	__skb_queue_head_init(&xmitq);
 
+	tipc_node_write_lock(n);
 	/* Initial node interval to value larger (10 seconds), then it will be
 	 * recalculated with link lowest tolerance
 	 */
-	tipc_node_read_lock(n);
 	n->keepalive_intv = 10000;
-	tipc_node_read_unlock(n);
+
+	remains = n->link_cnt;
+	if (n->dist_bulk) {
+		dist_bulk = true;
+		n->dist_bulk = false;
+	}
+	tipc_node_write_unlock_fast(n);
+	if (dist_bulk)
+		dist_rc = tipc_named_distribute(n->net, n->addr);
+
 	for (bearer_id = 0; remains && (bearer_id < MAX_BEARERS); bearer_id++) {
+		rc = 0;
 		tipc_node_read_lock(n);
 		le = &n->links[bearer_id];
 		if (le->link) {
@@ -839,11 +882,13 @@ static void tipc_node_timeout(struct timer_list *t)
 			tipc_node_calculate_timer(n, le->link);
 			rc = tipc_link_timeout(le->link, &xmitq);
 			spin_unlock_bh(&le->lock);
-			remains--;
+			if (dist_rc != -ENOBUFS)
+				remains--;
 		}
 		tipc_node_read_unlock(n);
 		tipc_bearer_xmit(n->net, bearer_id, &xmitq, &le->maddr, n);
-		if (rc & TIPC_LINK_DOWN_EVT)
+		/* Force the node down in case the redistribution failed */
+		if ((rc & TIPC_LINK_DOWN_EVT) || dist_rc == -ENOBUFS)
 			tipc_node_link_down(n, bearer_id, false);
 	}
 	mod_timer(&n->timer, jiffies + msecs_to_jiffies(n->keepalive_intv));
diff --git a/net/tipc/node.h b/net/tipc/node.h
index 775a8ec240fb..b312cefd4d41 100644
--- a/net/tipc/node.h
+++ b/net/tipc/node.h
@@ -130,4 +130,5 @@ int tipc_nl_node_set_key(struct sk_buff *skb, struct genl_info *info);
 int tipc_nl_node_flush_key(struct sk_buff *skb, struct genl_info *info);
 #endif
 void tipc_node_pre_cleanup_net(struct net *exit_net);
+void tipc_node_set_dist_bulk(struct net *net, u32 addr);
 #endif
diff --git a/net/tipc/socket.c b/net/tipc/socket.c
index 65cc0c4964b3..4f204bdcaf21 100644
--- a/net/tipc/socket.c
+++ b/net/tipc/socket.c
@@ -2929,7 +2929,7 @@ static int tipc_sk_publish(struct tipc_sock *tsk, struct tipc_uaddr *ua)
 		return -EADDRINUSE;
 	skaddr.ref = tsk->portid;
 	skaddr.node = tipc_own_addr(net);
-	p = tipc_nametbl_publish(net, ua, &skaddr, key);
+	p = tipc_nametbl_publish(net, ua, &skaddr, key, NULL);
 	if (unlikely(!p))
 		return -EINVAL;
 
-- 
2.53.0

Re: [PATCH net 1/2] tipc: fix self-deadlock in tipc_named_node_up

[netdev-bot+sashiko]

Thank you for your contribution! Sashiko AI review found 4 potential
issue(s) to consider.

Critical: 0 · High: 0 (1) · Medium: 2 (2) · Low: 2

If you believe none of the issues are legitimate, please reply to each
issue of Medium severity or higher and explain why the AI is wrong.
If any of them are legitimate and you plan to respin, please let patchwork
know by including "pw-bot: cr" as a separate line at the end of your reply
(one such reply per series is enough).

- [Medium] At the target commit, the -ENOBUFS recovery for bulk name
  distribution can reset a link other than the one that failed.
- [Medium] tipc: leak of cloned skbs in xmitq on -ENOBUFS in
  tipc_node_distribute()
- [Low] The commit message leaves out several behaviour changes and
  misdescribes the new helper.
- [Low] The new kernel-doc for tipc_node_distribute() lists the return
  value as `-ENOBUF` (net/tipc/node.c:1798).

Pre-existing issues:
- [High] named_distribute() runs `hdr = buf_msg(skb_peek_tail(list));`
  (name_distr.c:191) whether or not anything was queued.
- [Medium] At the target commit named_distribute() is still `static
  void`.
- [Medium] This problem wasn't introduced by this patch, but it is copied
  into the new code.

--

tipc: fix self-deadlock in tipc_named_node_up

This patch removes nt->cluster_scope_lock and uses tn->nametbl_lock to
protect the cluster_scope list. That breaks a self-deadlock in which
tipc_named_node_up() held the read lock across tipc_node_xmit() ->
tipc_node_link_down() -> tipc_nametbl_withdraw().

It adds tipc_node_distribute(), which transmits like tipc_node_xmit() but
does not reset the link, plus a shared helper, tipc_node_build_list().
tipc_named_node_up() now returns an error code, and
tipc_node_write_unlock() resets the link when that code is -ENOBUFS.

> Fix this issue by:
> - Removing cluster_scope_lock, which protects the name table's cluster_scope
>   list. Use nametbl_lock instead to protect the list against races when multiple
>   threads concurrently call tipc_named_publish(), tipc_named_withdraw(),
>   and tipc_named_node_up().
>
> - Introducing a new function, tipc_node_distribute(), which is similar to
>   tipc_node_xmit() but does not call tipc_node_link_down(). Both functions call
>   the newly introduced tipc_node_build_list(), which builds a list of messages
>   in sequence.

[Severity: Low]
Could the commit message cover a few more of the behaviour changes in this
patch?

The link reset on bulk failure is still there. It moved into
tipc_node_write_unlock():

	if (rc == -ENOBUFS)
		tipc_node_link_down(n, bearer_id, false);

tipc_named_node_up() now also maps -EMSGSIZE to -ENOBUFS, so an oversized
bulk now resets the link. Before this patch, tipc_node_xmit() returned
-EMSGSIZE and the void tipc_named_node_up() ignored it. The -ELINKCONG
mapping has a comment, but this one does not.

The message says nametbl_lock now protects the cluster_scope list. In fact
the whole transmit now runs under spin_lock_bh(&tn->nametbl_lock):
tipc_node_find(), the node read lock, le->lock, tipc_link_xmit(), and
tipc_bearer_xmit() all the way down to the media send_msg().

Also, tipc_node_build_list() does more than build a list of messages. It
either delivers to the peer namespace through tipc_lxc_xmit() or enqueues
on the link through tipc_link_xmit().

> diff --git a/net/tipc/name_distr.c b/net/tipc/name_distr.c
> index ba4f4906e13b7..a60f213ed1182 100644
> --- a/net/tipc/name_distr.c
> +++ b/net/tipc/name_distr.c

[ ... ]

> @@ -202,25 +198,39 @@ static void named_distribute(struct net *net, struct sk_buff_head *list,
>   * @net: the associated network namespace
>   * @dnode: destination node
>   * @capabilities: peer node's capabilities
> + *
> + * Return:
> + * * 0               - Success
> + * * -ENOBUFS        - No buffer space is available
> + * * -EHOSTUNREACH   - Destination node does not exist or cannot be reached
> + *
>   */
> -void tipc_named_node_up(struct net *net, u32 dnode, u16 capabilities)
> +int tipc_named_node_up(struct net *net, u32 dnode, u16 capabilities)
>  {
>  	struct name_table *nt = tipc_name_table(net);
>  	struct tipc_net *tn = tipc_net(net);
>  	struct sk_buff_head head;
> -	u16 seqno;
> +	int bearer_id = 0;
> +	int rc = 0;
>  
>  	__skb_queue_head_init(&head);
>  	spin_lock_bh(&tn->nametbl_lock);
>  	if (!(capabilities & TIPC_NAMED_BCAST))
>  		nt->rc_dests++;
> -	seqno = nt->snd_nxt;
> +	named_distribute(net, &head, dnode, &nt->cluster_scope, nt->snd_nxt);
> +	if (!skb_queue_empty(&head))
> +		rc = tipc_node_distribute(net, &head, dnode, &bearer_id, 0);

[Severity: High]
This is a pre-existing issue, but can named_distribute() dereference NULL
here when nt->cluster_scope is empty?

named_distribute() runs this whether or not it queued anything:

	hdr = buf_msg(skb_peek_tail(list));
	msg_set_last_bulk(hdr);

The list can be empty here. That happens if the TIPC_NOTIFY_NODE_UP event
is handled before tipc_net_finalize() has published the self
TIPC_NODE_STATE name, or if that publish failed. skb_peek_tail() then
returns NULL, and buf_msg(NULL) is dereferenced in softirq context.

A remote peer bringing up a link can trigger this. A syzbot report shows
the path tipc_rcv() -> tipc_node_write_unlock() -> tipc_named_node_up().

The new skb_queue_empty() check only runs after the dereference.

The follow-up patch "tipc: fix NULL pointer dereference in
named_distribute()" appears to fix this. It routes the call through
__tipc_named_distribute(), which makes sure the self publication is on
nt->cluster_scope, or defers the bulk, before named_distribute() runs.

[Severity: Medium]
This isn't a bug introduced by this patch, but the new kernel-doc says
tipc_named_node_up() returns -ENOBUFS when no buffer space is available.
Can that happen for allocation failures?

named_distribute() is still void. When named_prepare_buf() fails, it
returns without purging the partial list and without setting last_bulk or
the seqno:

			if (!skb) {
				pr_warn("Bulk publication failure\n");
				return;
			}

rc is only set from tipc_node_distribute(), so this case returns 0. The
new -ENOBUFS recovery in tipc_node_write_unlock() never runs. The peer's
tipc_named_dequeue() never sees last_bulk, so the peer's name table stays
stale.

The follow-up patch "tipc: fix NULL pointer dereference in
named_distribute()" seems to fix this. There, named_distribute() purges
the list and returns -ENOBUFS, and tipc_named_node_up() passes that
error up.

[ ... ]

> diff --git a/net/tipc/node.c b/net/tipc/node.c
> index 53b0993c2c233..b19d6652dc731 100644
> --- a/net/tipc/node.c
> +++ b/net/tipc/node.c

[ ... ]

> @@ -435,6 +436,9 @@ static void tipc_node_write_unlock(struct tipc_node *n)
>  		tipc_mon_peer_down(net, node, bearer_id);
>  		tipc_nametbl_withdraw(net, &ua, &sk, sk.ref);
>  	}
> +
> +	if (rc == -ENOBUFS)
> +		tipc_node_link_down(n, bearer_id, false);

[Severity: Medium]
Can this reset a different link from the one the bulk failed on?

Here bearer_id comes from n->link_id. It is saved before n->lock is
dropped:

	bearer_id = n->link_id & 0xffff;

The bulk is sent on whichever bearer tipc_node_build_list() picks later,
in a separate lock section:

	*bearer_id = n->active_links[selector & 1];

tipc_named_node_up() receives that bearer from tipc_node_distribute() in
its local bearer_id, then discards it and returns only rc.

Suppose a higher priority second link comes up in that window.
__tipc_node_link_up() puts the new bearer in both active slots, and the
bulk then fails on the new link. The old, healthy link is the one that
gets reset, and the bulk is never resent. The old tipc_node_xmit() path
reset the bearer it had actually used.

In the follow-up patch "tipc: fix NULL pointer dereference in
named_distribute()" this call becomes tipc_node_down(n), which brings
down every bearer. So the problem seems to be gone by the end of the
series.

[ ... ]

> @@ -1715,42 +1763,66 @@ int tipc_node_xmit(struct net *net, struct sk_buff_head *list,

[ ... ]

> +/**
> + * tipc_node_distribute() - distribute list of messages to specific destination
> + * @net: the applicable net namespace
> + * @list: chain of buffers containing message
> + * @dnode: address of destination node
> + * @bearer_id: id of the bearer
> + * @selector: a number used for deterministic link selection
> + *
> + * Return:
> + * * 0              - Success
> + * * -ELINKCONG     - Link is congested
> + * * -EHOSTUNREACH  - Host is not reachable
> + * * -EMSGSIZE      - Message's size is too big
> + * * -ENOBUF        - No buffer space is available

[Severity: Low]
This isn't a bug, but -ENOBUF should be -ENOBUFS. The typo seems to have
been copied from the tipc_node_xmit() comment, and it is still there at
the end of the series.

> + *
> + */
> +int tipc_node_distribute(struct net *net, struct sk_buff_head *list,
> +			 u32 dnode, int *bearer_id, int selector)
> +{

[ ... ]

>  	__skb_queue_head_init(&xmitq);
> -	le = &n->links[bearer_id];
> -	spin_lock_bh(&le->lock);
> -	rc = tipc_link_xmit(le->link, list, &xmitq);
> -	spin_unlock_bh(&le->lock);
> -	tipc_node_read_unlock(n);
> +	rc = tipc_node_build_list(n, list, &xmitq, selector, &maddr,
> +				  bearer_id, &lxc);
> +	if (lxc) {
> +		tipc_node_put(n);
> +		return 0;
> +	}
>  
> -	if (unlikely(rc == -ENOBUFS))
> -		tipc_node_link_down(n, bearer_id, false);
> -	else
> -		tipc_bearer_xmit(net, bearer_id, &xmitq, &le->maddr, n);
> +	if (likely(rc != -ENOBUFS))
> +		tipc_bearer_xmit(net, *bearer_id, &xmitq, &maddr, n);

[Severity: Medium]
This is a pre-existing issue, but can the clones already queued on
xmitq leak when tipc_link_xmit() returns -ENOBUFS here?

tipc_link_xmit() takes each skb off list, puts it on l->transmq and
puts a skb_clone() of it on the caller's xmitq. Suppose skb_clone()
fails on a later skb in the list. It frees that skb, purges list and
returns -ENOBUFS, but the clones from the earlier iterations stay on
xmitq.

On -ENOBUFS, tipc_node_distribute() skips tipc_bearer_xmit() and
returns. Nothing ever purges the stack-local xmitq, so those clones
are lost.

The refactored tipc_node_xmit() has the same problem. It calls
tipc_node_link_down() instead. That only cleans up the link's own
queues, not the caller's xmitq.

This needs a multi-buffer list, for example a bulk from
named_distribute() when nt->cluster_scope spans more than one MTU. It
also needs a GFP_ATOMIC failure in softirq context, so it is most
likely under memory pressure.

Both callers now go through tipc_node_build_list(). Would it make
sense to purge xmitq there on -ENOBUFS? Or should tipc_link_xmit()
clean up the clones it queued before returning the error?

[Severity: Medium]
This is a pre-existing issue, but are the skb clones on xmitq leaked when
rc is -ENOBUFS?

tipc_link_xmit() clones each skb onto xmitq. If skb_clone() fails on a
later skb, it returns -ENOBUFS and the earlier clones stay queued:

			_skb = skb_clone(skb, GFP_ATOMIC);
			if (!_skb) {
				kfree_skb(skb);
				__skb_queue_purge(list);
				return -ENOBUFS;
			}

On -ENOBUFS this function skips tipc_bearer_xmit() and never purges the
stack-local xmitq. The refactored tipc_node_xmit() does the same, as did
the original version.

Would a __skb_queue_purge(xmitq) on -ENOBUFS in tipc_node_build_list()
cover both callers? The leak is still present at the end of the series.

-- 
Sashiko AI review · https://netdev-ai.bots.linux.dev/sashiko/#/patchset/20261028033722.125477-1-tung.quang.nguyen%40est.tech

Re: [PATCH net 2/2] tipc: fix NULL pointer dereference in named_distribute()

[netdev-bot+sashiko]

Thank you for your contribution! Sashiko AI review found 6 potential
issue(s) to consider.

Critical: 1 · High: 0 · Medium: 3 (2) · Low: 2

If you believe none of the issues are legitimate, please reply to each
issue of Medium severity or higher and explain why the AI is wrong.
If any of them are legitimate and you plan to respin, please let patchwork
know by including "pw-bot: cr" as a separate line at the end of your reply
(one such reply per series is enough).

- [Critical] The fix is incomplete: a TIPC peer can still trigger the
  NULL dereference in named_distribute() (buf_msg(skb_peek_tail(list))…
- [Medium] The commit message lists "three major changes", all internal
  refactoring.
- [Medium] The new deferred-bulk flag n->dist_bulk can go stale and cause
  a second bulk distribution within one node session.
- [Medium] tipc: race in tipc_net_finalize() causes self-node publish to
  fail with -EEXIST
- [Low] The comment on the tn->nt_stop check in __tipc_named_distribute()
  says it protects against the name table having been deleted at…
- [Low] The new comment in tipc_service_delete() (net/tipc/name_table.c)
  says "tipc_nametbl_withdraw() does not delete 'p->binding_node' with…

Pre-existing issues:
- [Medium] tipc_named_reinit(), called from tipc_net_finalize(), sets
  nt->rc_dests = 0 unconditionally.
- [Medium] tipc_node_distribute() can leak SKBs, and the new timer path
  tipc_named_distribute() now reaches it too.

--

tipc: fix NULL pointer dereference in named_distribute()

This patch fixes a syzbot-reported NULL dereference in named_distribute().
The cause is a race between tipc_net_finalize() and the node-up bulk
distribution. The patch inserts the node's self-address TIPC_NODE_STATE
publication on demand from node-up or the node timer. It defers the bulk to
the node timer when the address is not set yet, and it adds error
propagation for allocation failures.

Closes: https://syzkaller.appspot.com/bug?extid=64197f7577a09686ffdb

> To facilitate the implementation, three major changes are made:
> - Move the sc->lock acquisition out of tipc_service_insert_publ() and into its
>   callers.
> - Add an output parameter (*err) to tipc_nametbl_publish() to detect memory
>   allocation errors.
> - Add three fields (nt_stop, nt_self_node_exist, and node_addr_set) to struct
>   tipc_net to synchronize the insertion and deletion of the node's self-address
>   publication.

[Severity: Medium]

These three items read as internal refactoring. The patch also changes
behaviour in several ways that the commit message does not mention. Should
these be described, or split into separate patches?

tipc_update_nametbl() now drops every peer PUBLICATION or WITHDRAWAL of
type TIPC_NODE_STATE whose range equals our address. It also logs "Can't
update duplicate node-up pub".

In tipc_node_write_unlock(), -ENOBUFS from tipc_named_node_up() now takes
down every link to the peer through the new tipc_node_down(). In the
parent commit ("tipc: fix self-deadlock in tipc_named_node_up") it only
called tipc_node_link_down(n, bearer_id, false).

An allocation failure while publishing the node-scope TIPC_LINK_STATE
binding on LINK_UP used to be ignored. It now forces that link down, which
seems unrelated to the named_distribute() race.

tipc_node_timeout() now forces every link to the peer down when the
deferred bulk redistribution returns -ENOBUFS.

Separately, can this new branch in tipc_nametbl_publish() ever run?

	if (!skb && p->scope == TIPC_CLUSTER_SCOPE && err)
		*err = -ENOBUFS;

The only caller that passes a non-NULL err is tipc_node_write_unlock(), and
it publishes TIPC_LINK_STATE with TIPC_NODE_SCOPE. tipc_net_finalize() and
tipc_sk_publish() both pass NULL.

> diff --git a/net/tipc/name_distr.c b/net/tipc/name_distr.c
> index a60f213ed1182..345bd767983e9 100644
> --- a/net/tipc/name_distr.c
> +++ b/net/tipc/name_distr.c

[ ... ]

> @@ -191,6 +197,94 @@ static void named_distribute(struct net *net, struct sk_buff_head *list,
>  	hdr = buf_msg(skb_peek_tail(list));
>  	msg_set_last_bulk(hdr);
>  	msg_set_named_seqno(hdr, seqno);
> +
> +	return 0;
> +}

[ ... ]

> +static int __tipc_named_distribute(struct net *net, u32 dnode,
> +				   struct sk_buff_head *head)
> +{
> +	struct name_table *nt = tipc_name_table(net);
> +	struct tipc_net *tn = tipc_net(net);
> +
> +	/* Name table has been deleted after namespace teardown or
> +	 * TIPC module exit.
> +	 */
> +	if (unlikely(tn->nt_stop))
> +		return 0;

[Severity: Low]

Can this check protect against a deleted name table? tipc_named_node_up()
already dereferences nt before __tipc_named_distribute() reads nt_stop:

	if (!(capabilities & TIPC_NAMED_BCAST))
		nt->rc_dests++;
	rc = __tipc_named_distribute(net, dnode, &head);

The scenario also does not look reachable. tipc_exit_net() runs
tipc_net_stop() first, which goes through
tipc_node_stop()->tipc_node_delete()->timer_delete_sync(). It then runs
cancel_work_sync(&tn->work), and only after that tipc_nametbl_stop().
tipc_nametbl_stop() sets nt_stop just before kfree_rcu(nt).

The only other caller of tipc_nametbl_stop() is the tipc_init_net() error
path, and no nodes exist at that point.

Is nt_stop needed at all? The commit message says it helps to "synchronize
the insertion and deletion of the node's self-address publication", which
does not seem accurate either.

> +
> +	/* Node's self-address is not set yet */
> +	if (!atomic_read(&tn->node_addr_set)) {
> +		/* Defer the distribution to node's timer */
> +		tipc_node_set_dist_bulk(net, dnode);
> +		return 0;
> +	}

[ ... ]

> +int tipc_named_distribute(struct net *net, u32 dnode)
> +{
> +	struct tipc_net *tn = tipc_net(net);
> +	struct sk_buff_head head;
> +	int bearer_id = 0;
> +	int rc = 0;
> +
> +	__skb_queue_head_init(&head);
> +	spin_lock_bh(&tn->nametbl_lock);
> +	rc = __tipc_named_distribute(net, dnode, &head);
> +	if (!rc && !skb_queue_empty(&head))
> +		rc = tipc_node_distribute(net, &head, dnode, &bearer_id, 0);

[Severity: Medium]

This is a pre-existing issue, not introduced by this patch, but the new
timer path through tipc_named_distribute() now reaches it as well. Can
tipc_node_distribute() leak skbs on its local xmitq?

When tipc_link_xmit() sends a multi-packet bulk, earlier packets may
already be cloned onto xmitq when a later skb_clone() fails:

	_skb = skb_clone(skb, GFP_ATOMIC);
	if (!_skb) {
		kfree_skb(skb);
		__skb_queue_purge(list);
		return -ENOBUFS;
	}

tipc_node_distribute() then skips the transmit and never purges xmitq:

	if (likely(rc != -ENOBUFS))
		tipc_bearer_xmit(net, *bearer_id, &xmitq, &maddr, n);

	tipc_node_put(n);

tipc_node_xmit() had the same pattern, and tipc_named_node_up() used it
before. tipc_node_distribute() was copied from it in the preceding commit
"tipc: fix self-deadlock in tipc_named_node_up".

> +	spin_unlock_bh(&tn->nametbl_lock);

[ ... ]

> @@ -426,5 +531,6 @@ void tipc_named_reinit(struct net *net)
>  	list_for_each_entry_rcu(p, &nt->cluster_scope, binding_node)
>  		p->sk.node = self;
>  	nt->rc_dests = 0;
> +	tn->nt_stop = false;
>  	spin_unlock_bh(&tn->nametbl_lock);
>  }

[Severity: Medium]

This is a pre-existing issue, not introduced by this patch. The patch now
explicitly supports node-up before finalize, though, so it seems relevant
here.

tipc_net_finalize() calls tipc_named_reinit(), which resets nt->rc_dests to
0 unconditionally. That wipes out the increment tipc_named_node_up() has
already made for any peer without TIPC_NAMED_BCAST that came up before
finalize:

tipc_named_node_up()
  nt->rc_dests++;
  __tipc_named_distribute()
    tipc_node_set_dist_bulk()     <- only the bulk is deferred
...
tipc_net_finalize()
  tipc_named_reinit()
    nt->rc_dests = 0;

Would later publications then see rc_dests == 0 and go out over the
broadcast link? Legacy peers do not accept name distribution that way.

When such a peer goes down, tipc_publ_notify() does nt->rc_dests--, which
wraps the u32 to 0xFFFFFFFF. tipc_node_broadcast() receives that as int -1
and stays in replicast from then on.

> diff --git a/net/tipc/name_table.c b/net/tipc/name_table.c
> index 8924bdb6a443f..3bcb8762bc9cc 100644
> --- a/net/tipc/name_table.c
> +++ b/net/tipc/name_table.c

[ ... ]

> +int tipc_nametb_insert_self_node_pub(struct net *net)
> +{

[ ... ]

> +	spin_lock_bh(&sc->lock);
> +	sr = tipc_service_find_range(sc, &ua);
> +	/* 1. tipc_net_finalize_work() is not scheduled because of namespace
> +	 *    teardown.
> +	 * 2. Or tipc_net_finalize() ---> tipc_nametbl_publish() has failed
> +	 *    to insert node self address publication into nt->cluster_scope
> +	 *    due to memory allocation failure.
> +	 * 3. Or tipc_net_finalize() ---> tipc_nametbl_publish() has not
> +	 *    executed yet.
> +	 */
> +	if (!sr) {

[ ... ]

> +exit:
> +	spin_unlock_bh(&sc->lock);
> +	return 0;
> +}

[Severity: Medium]
Is there a window between setting node_addr_set and the
tipc_nametbl_publish() call that follows it? node_addr_set is written
here without tn->nametbl_lock, but __tipc_named_distribute() reads it
under that lock. A node-up, or the node timer, can run in between. That
could be on another CPU, or in softirq on this one. It then does the
self-node insert first:

CPU0                          CPU1
tipc_net_finalize()
  atomic_set(node_addr_set)
                              tipc_named_node_up() or
                              tipc_node_timeout()
                                __tipc_named_distribute()
                                  tipc_nametb_insert_self_node_pub()
                                  tn->nt_self_node_exist = true;
                                  named_distribute()  <- dnode only
  tipc_nametbl_publish()
    tipc_nametbl_insert_publ()
      tipc_service_insert_publ()  <- duplicate, -EEXIST

tipc_nametb_insert_self_node_pub() builds the publication with the same
key, sk.ref and sk.node that tipc_net_finalize() passes. The duplicate
check in tipc_service_insert_publ() therefore rejects the second insert.
That prints the non-ratelimited "Failed to bind to" pr_warn() for what
is really an expected race. tipc_nametbl_publish() then returns NULL
without calling tipc_named_publish() or tipc_node_broadcast().

In that case the TIPC_NODE_STATE binding reaches only dnode, through
its bulk. Peers that are already up never get the broadcast that
finalize normally sends. They depend on their own deferred dist_bulk
being processed later. Is that intended?

The new comment above the call says the publication is redistributed
if tipc_nametbl_publish() fails. The -EEXIST case is not really a
failure, though, and it still warns.

Would it work to set node_addr_set only after tipc_nametbl_publish()
has returned? Alternatively, tipc_net_finalize() could insert the
publication under tn->nametbl_lock and set nt_self_node_exist there, so
both paths agree on which one owns the insert.

[Severity: Critical]

Can a [self,self] TIPC_NODE_STATE range that holds only a peer's entry make
this function return 0 without adding anything to nt->cluster_scope? The
function only checks that the range exists. It does not check that the
range holds this node's own publication, or that anything is linked on
nt->cluster_scope.

The new filter in tipc_update_nametbl() is skipped while tn->node_addr is
still 0:

	if (unlikely(type == TIPC_NODE_STATE && tn->node_addr &&
		     lower == tn->node_addr && upper == tn->node_addr)) {

During the address trial, tipc_disc_addr_trial_msg() schedules tn->work but
lets link setup continue while self == 0. The peer can see our trial
address, because we put it in prevnode in our discovery messages.
tipc_named_rcv() takes node = msg_orignode(hdr) without checking it against
the link peer.

So a peer could send an item like {type=TIPC_NODE_STATE,
lower=upper=trial_addr, port=0, key=trial_addr} with orignode=trial_addr.
orignode=0 would also work, because the !_p->sk.node test in
tipc_service_insert_publ() treats it as a wildcard. The sequence would then
be:

tipc_net_finalize()
  atomic_set(&tn->node_addr_set, 1);
  tipc_nametbl_publish(net, &ua, &sk, addr, NULL)
    tipc_service_insert_publ()   <- duplicate, -EEXIST, ignored

tipc_named_node_up() or tipc_node_timeout()->tipc_named_distribute()
  __tipc_named_distribute()
    tipc_nametb_insert_self_node_pub()   <- range found, returns 0
    tn->nt_self_node_exist = true;
    named_distribute(net, head, dnode, &nt->cluster_scope, ...)
      hdr = buf_msg(skb_peek_tail(list));   <- list is empty

With no cluster-scope user bindings, nt->cluster_scope is empty at this
point. The topology server binds with TIPC_NODE_SCOPE, so it does not add
anything. Would this hit the same NULL dereference in buf_msg() that the
patch sets out to fix, in softirq or timer context?

Even with an honest orignode, a planted [self,self] range still makes this
helper skip the insert. That reopens the window between
atomic_set(&tn->node_addr_set, 1) and tipc_nametbl_publish() in
tipc_net_finalize().

Would it be safer to check that our own publication is actually on
nt->cluster_scope? It might also help to have named_distribute() handle an
empty list, for example by returning early when skb_peek_tail(list) is
NULL.

[ ... ]

> @@ -940,6 +1043,15 @@ static void tipc_service_delete(struct net *net, struct tipc_service *sc)
>  	rbtree_postorder_for_each_entry_safe(sr, tmpr, &sc->ranges, tree_node) {
>  		list_for_each_entry_safe(p, tmp, &sr->all_publ, all_publ) {
>  			tipc_service_remove_publ(sr, &p->sk, p->key);
> +			/* tipc_nametbl_withdraw() does not delete
> +			 * 'p->binding_node' with type TIPC_NODE_STATE and
> +			 * range [tn->node_addr, tn->node_addr] from
> +			 * 'tn->cluster_scope'. So, delete it here.
> +			 */

[Severity: Low]

Is this comment accurate? tipc_nametbl_withdraw() calls
tipc_named_withdraw() for each publication it removes, and
tipc_named_withdraw() begins with an unconditional:

	list_del(&p->binding_node);

The self-node publication is still linked here only because nothing ever
withdraws it. tipc_net_stop() just calls tipc_bearer_stop() and
tipc_node_stop().

cluster_scope is also a member of struct name_table, not struct tipc_net,
so 'tn->cluster_scope' should probably read 'nt->cluster_scope'.

If a later change follows this comment and adds a withdraw call, would this
list_del() then remove the same entry twice?

> +			if (p->sr.type == TIPC_NODE_STATE &&
> +			    p->sr.lower == tn->node_addr &&
> +			    p->sr.upper == tn->node_addr)
> +				list_del(&p->binding_node);
>  			kfree_rcu(p, rcu);
>  		}

[ ... ]

> diff --git a/net/tipc/node.c b/net/tipc/node.c
> index b19d6652dc731..f566472a411fc 100644
> --- a/net/tipc/node.c
> +++ b/net/tipc/node.c

[ ... ]

> @@ -420,6 +430,8 @@ static void tipc_node_write_unlock(struct tipc_node *n)
>  	n->action_flags &= ~(TIPC_NOTIFY_NODE_DOWN | TIPC_NOTIFY_NODE_UP |
>  			     TIPC_NOTIFY_LINK_DOWN | TIPC_NOTIFY_LINK_UP);
>  
> +	if (flags & TIPC_NOTIFY_NODE_DOWN)
> +		n->dist_bulk = false;
>  	write_unlock_bh(&n->lock);

[ ... ]

> +void tipc_node_set_dist_bulk(struct net *net, u32 addr)
> +{
> +	struct tipc_node *node;
> +
> +	node = tipc_node_find(net, addr);
> +	if (node) {
> +		tipc_node_write_lock(node);
> +		node->dist_bulk = true;
> +		tipc_node_write_unlock_fast(node);
> +		tipc_node_put(node);
> +	}
> +}

[Severity: Medium]

Can n->dist_bulk go stale and trigger a second bulk within one node
session?

tipc_node_write_unlock() clears the flag on TIPC_NOTIFY_NODE_DOWN under
n->lock. tipc_named_node_up() runs only after that lock is dropped, and
tipc_node_set_dist_bulk() sets the flag without checking whether the node
has gone down in the meantime:

CPU0                                  CPU1
tipc_node_write_unlock() (NODE_UP)
  write_unlock_bh(&n->lock);
                                      link failure, NODE_DOWN
                                        n->dist_bulk = false;
  tipc_named_node_up()
    __tipc_named_distribute()
      node_addr_set == 0
      tipc_node_set_dist_bulk()
        node->dist_bulk = true;

Suppose the peer comes back after finalize has set node_addr_set. The new
NODE_UP sends a normal bulk right away. Later the node timer finds the
stale flag and sends a second one.

tipc_node_timeout() has a similar window. It reads and clears the flag
under n->lock, but calls tipc_named_distribute() after unlocking without
checking again.

On the receiver, tipc_named_dequeue() advances rcv_nxt to the extra bulk's
seqno and drops queued messages with a lower seqno:

	if (msg_is_last_bulk(hdr)) {
		*rcv_nxt = seqno;
		*open = true;
	}
	...
	if (less(seqno, *rcv_nxt)) {
		__skb_unlink(skb, namedq);
		kfree_skb(skb);

Could a withdrawal that has not been applied yet be lost this way, leaving
a stale binding on the peer? The duplicate items would also trigger the
non-ratelimited pr_warn() in tipc_service_insert_publ().

[ ... ]

-- 
Sashiko AI review · https://netdev-ai.bots.linux.dev/sashiko/#/patchset/20261028033722.125477-1-tung.quang.nguyen%40est.tech

Re: [PATCH net 2/2] tipc: fix NULL pointer dereference in named_distribute()

[Paolo Abeni]

On 10/28/26 04:36, Tung Nguyen wrote:
> +/**
> + * tipc_nametb_insert_self_node_pub - insert publication of node's self address
> + * @net: network namespace
> + *
> + * Return:
> + * * 0          - Success
> + * * -ENOBUFS   - No buffer space is available
> + *
> + */
> +int tipc_nametb_insert_self_node_pub(struct net *net)
> +{
> +	struct name_table *nt = tipc_name_table(net);
> +	struct tipc_net *tn = tipc_net(net);
> +	struct tipc_socket_addr sk;
> +	struct service_range *sr;
> +	struct tipc_service *sc;
> +	bool sc_created = false;
> +	struct publication *p;
> +	struct tipc_uaddr ua;
> +	int err = 0;
> +
> +	sk.ref = 0;
> +	sk.node = tn->node_addr;
> +	tipc_uaddr(&ua, TIPC_SERVICE_RANGE, TIPC_CLUSTER_SCOPE,
> +		   TIPC_NODE_STATE, tn->node_addr, tn->node_addr);
> +	sc = tipc_service_find(net, &ua);
> +	if (!sc) {
> +		sc = tipc_service_create(net, &ua);
> +		if (!sc)
> +			return -ENOBUFS;
> +		sc_created = true;
> +	}
> +
> +	spin_lock_bh(&sc->lock);
> +	sr = tipc_service_find_range(sc, &ua);
> +	/* 1. tipc_net_finalize_work() is not scheduled because of namespace
> +	 *    teardown.
> +	 * 2. Or tipc_net_finalize() ---> tipc_nametbl_publish() has failed
> +	 *    to insert node self address publication into nt->cluster_scope
> +	 *    due to memory allocation failure.
> +	 * 3. Or tipc_net_finalize() ---> tipc_nametbl_publish() has not
> +	 *    executed yet.
> +	 */
> +	if (!sr) {
> +		p = tipc_publ_create(&ua, &sk, tn->node_addr);
> +		if (!p)
> +			goto error;
> +
> +		if (!tipc_service_insert_publ(net, sc, p, &err)) {
> +			kfree(p);
> +			goto error;
> +		}
> +
> +		list_add_tail(&p->binding_node, &nt->cluster_scope);
> +		goto exit;
> +error:
> +		if (sc_created) {

The above is a bit ugly. AFAICS you avoid the unneeded label `exit`
moving this error path, label included, after the `return 0;` statement.

Also sashiko complains look legit.

/P