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

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

[Tung Nguyen]

Patch 1 fixes a memory leak in tipc_node_xmit().

Patch 2 fixes a dangling link in the cluster_scope list during network
namespace teardown.

Patch 3 fixes incorrect initialization of the replicast node counter.

Patch 4 fixes a missing link-up event when memory allocation fails.

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

Patch 6 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.

Changes in v2:
- Add patches 1–4 to fix pre-existing issues discovered while reviewing
  patch 6.
- Patch 5: Correct the bearer id that needs to be reset.
- Patch 6: Set node->dist_bulk to true when the node comes up, and
  remove an unnecessary publication insertion in tipc_update_nametbl().

Tung Nguyen (6):
  tipc: fix memory leak in tipc_node_xmit()
  tipc: fix dangling link in cluster_scope list in tipc_service_delete()
  tipc: fix incorrect initialization of replicast node counter
  tipc: fix missing link-up event on memory allocation failure
  tipc: fix self-deadlock in tipc_named_node_up
  tipc: fix NULL pointer dereference in named_distribute()

 net/tipc/core.c       |   1 +
 net/tipc/core.h       |   2 +
 net/tipc/name_distr.c | 124 +++++++++++++++++++++----
 net/tipc/name_distr.h |   4 +-
 net/tipc/name_table.c | 133 +++++++++++++++++++++++----
 net/tipc/name_table.h |  11 ++-
 net/tipc/net.c        |  19 +++-
 net/tipc/node.c       | 208 +++++++++++++++++++++++++++++++++---------
 net/tipc/node.h       |   3 +
 net/tipc/socket.c     |   2 +-
 10 files changed, 421 insertions(+), 86 deletions(-)

-- 
2.53.0

[PATCH net v2 1/6] tipc: fix memory leak in tipc_node_xmit()

[Tung Nguyen]

The following memory leak can occur, as identified during code inspection:

tipc_node_xmit()
 tipc_link_xmit(le->link, list, &xmitq)
  ...
  __skb_queue_purge(list);
  return -ENOBUFS;

tipc_link_xmit() purges list but does not purge xmitq. tipc_node_xmit() does not
purge xmitq either, resulting in a memory leak.

Fix this issue by purging xmitq in tipc_node_xmit() when tipc_link_xmit()
returns -ENOBUFS.

Fixes: 4952cd3e7b47 ("tipc: refactor node xmit and fix memory leaks")
Signed-off-by: Tung Nguyen <tung.quang.nguyen@est.tech>
---
 net/tipc/node.c | 7 +++++--
 1 file changed, 5 insertions(+), 2 deletions(-)

diff --git a/net/tipc/node.c b/net/tipc/node.c
index 53b0993c2c23..d84b0274fefc 100644
--- a/net/tipc/node.c
+++ b/net/tipc/node.c
@@ -1747,10 +1747,13 @@ int tipc_node_xmit(struct net *net, struct sk_buff_head *list,
 	spin_unlock_bh(&le->lock);
 	tipc_node_read_unlock(n);
 
-	if (unlikely(rc == -ENOBUFS))
+	if (unlikely(rc == -ENOBUFS)) {
+		if (!skb_queue_empty(&xmitq))
+			__skb_queue_purge(&xmitq);
 		tipc_node_link_down(n, bearer_id, false);
-	else
+	} else {
 		tipc_bearer_xmit(net, bearer_id, &xmitq, &le->maddr, n);
+	}
 
 	tipc_node_put(n);
 
-- 
2.53.0

[PATCH net v2 2/6] tipc: fix dangling link in cluster_scope list in tipc_service_delete()

[Tung Nguyen]

The TIPC name table uses the cluster_scope list to store the
following publications:

- User publications with cluster scope from the local node:

  __tipc_bind()
   tipc_sk_publish()
    tipc_nametbl_publish()
     tipc_named_publish()
      list_add_tail(&p->binding_node, &nt->cluster_scope)
    list_add(&p->binding_sock, &tsk->publications)

- Protocol publications of type TIPC_NODE_STATE (0) with cluster
  scope from the local node:

  tipc_net_finalize()
   tipc_nametbl_publish()
    tipc_named_publish()
     list_add_tail(&p->binding_node, &nt->cluster_scope)

When a network namespace is destroyed:

- User publications with cluster scope from the local node are removed
  through the following call path:

  tipc_release()
    tipc_sk_withdraw()
     list_for_each_entry_safe(p, safe, &tsk->publications, binding_sock)
      tipc_nametbl_withdraw()
       tipc_named_withdraw()
        list_del(&p->binding_node)

- However, code inspection revealed that tipc_service_delete() does not
  remove self-node TIPC_NODE_STATE publication from the cluster_scope list,
  leaving dangling list entries:

   tipc_service_delete()
    list_for_each_entry_safe()
     tipc_service_remove_publ()
      kfree_rcu(p, rcu);

Fix this by calling list_del() before freeing TIPC_NODE_STATE
publications in tipc_service_delete().

Fixes: 1bb8dce57f4d ("tipc: fix memory leak during module removal")
Signed-off-by: Tung Nguyen <tung.quang.nguyen@est.tech>
---
 net/tipc/name_table.c | 8 ++++++++
 1 file changed, 8 insertions(+)

diff --git a/net/tipc/name_table.c b/net/tipc/name_table.c
index 6fda36ab1766..73b032ba2f8d 100644
--- a/net/tipc/name_table.c
+++ b/net/tipc/name_table.c
@@ -934,6 +934,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;
 
@@ -941,6 +942,13 @@ 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);
+			if (p->sr.type == TIPC_NODE_STATE &&
+			    p->sr.lower == tn->node_addr &&
+			    p->sr.upper == tn->node_addr &&
+			    p->key == tn->node_addr &&
+			    p->sk.ref == 0 && p->sk.node == tn->node_addr)
+				list_del(&p->binding_node);
+
 			kfree_rcu(p, rcu);
 		}
 		rb_erase_augmented(&sr->tree_node, &sc->ranges, &sr_callbacks);
-- 
2.53.0

[PATCH net v2 3/6] tipc: fix incorrect initialization of replicast node counter

[Tung Nguyen]

The TIPC name table uses rc_dests to track the number of nodes
available for replicast transmission. This counter is used to decide
whether to use broadcast or replicast links when sending bulk messages.

However, code inspection revealed a race between tipc_net_finalize(),
which initializes rc_dests, and tipc_named_node_up(), which
increments it, as illustrated below:

Thread 1 (tipc_net_finalize) | Thread 2 (tipc_named_node_up)
                             | ...
                             | nt->rc_dests++;
...                          |
 tipc_named_reinit()         |
   nt->rc_dests = 0;         |

If tipc_named_reinit() (Thread 1) executes after nt->rc_dests++
(Thread 2), the counter will be reset to zero. As a result, two
unexpected behaviors can occur:

1. User publications are sent via the broadcast link instead of the
   replicast link:

   tipc_sk_publish()
    tipc_nametbl_publish()
     tipc_node_broadcast()
      if (!rc_dests && tipc_bcast_get_mode(net) != BCLINK_MODE_RCAST)
       tipc_bcast_xmit()

2. nt->rc_dests can underflow to -1 when the node goes down. As a
   result, publications are sent via the replicast link even though no
   replicast nodes are available:

   tipc_publ_notify()
    nt->rc_dests--;

Fix this issue by moving initialization of nt->rc_dests from
tipc_named_reinit() to tipc_nametbl_init().

Fixes: cad2929dc432 ("tipc: update a binding service via broadcast")
Signed-off-by: Tung Nguyen <tung.quang.nguyen@est.tech>
---
 net/tipc/name_distr.c | 1 -
 net/tipc/name_table.c | 1 +
 2 files changed, 1 insertion(+), 1 deletion(-)

diff --git a/net/tipc/name_distr.c b/net/tipc/name_distr.c
index ba4f4906e13b..1dfe8475a75d 100644
--- a/net/tipc/name_distr.c
+++ b/net/tipc/name_distr.c
@@ -415,6 +415,5 @@ void tipc_named_reinit(struct net *net)
 		p->sk.node = self;
 	list_for_each_entry_rcu(p, &nt->cluster_scope, binding_node)
 		p->sk.node = self;
-	nt->rc_dests = 0;
 	spin_unlock_bh(&tn->nametbl_lock);
 }
diff --git a/net/tipc/name_table.c b/net/tipc/name_table.c
index 73b032ba2f8d..962b82956032 100644
--- a/net/tipc/name_table.c
+++ b/net/tipc/name_table.c
@@ -922,6 +922,7 @@ 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);
+	nt->rc_dests = 0;
 	tn->nametbl = nt;
 	spin_lock_init(&tn->nametbl_lock);
 	return 0;
-- 
2.53.0

[PATCH net v2 4/6] tipc: fix missing link-up event on memory allocation failure

[Tung Nguyen]

When a new TIPC link comes up, a protocol publication of type
TIPC_LINK_STATE (0) is created and inserted into the name table
through the following call path:

tipc_node_write_unlock()
 tipc_nametbl_publish()
  tipc_nametbl_insert_publ()

User applications subscribing to this TIPC_LINK_STATE event will
be notified when the link comes up. However, code inspection
revealed that if tipc_nametbl_insert_publ() fails due to a memory
allocation error, the TIPC_LINK_STATE publication is not inserted
into the name table. As a result, user applications are never
notified of the link-up event.

Fix this issue by checking the error code returned through the
output parameter of tipc_nametbl_publish(). If the error indicates
a memory allocation failure, reset the link so that the
TIPC_LINK_STATE publication can be retried.

Fixes: b97bf3fd8f6a ("[TIPC] Initial merge")
Signed-off-by: Tung Nguyen <tung.quang.nguyen@est.tech>
---
 net/tipc/name_distr.c |  2 +-
 net/tipc/name_table.c | 43 ++++++++++++++++++++++++++++++++++---------
 net/tipc/name_table.h |  5 +++--
 net/tipc/net.c        |  2 +-
 net/tipc/node.c       |  6 +++++-
 net/tipc/socket.c     |  2 +-
 6 files changed, 45 insertions(+), 15 deletions(-)

diff --git a/net/tipc/name_distr.c b/net/tipc/name_distr.c
index 1dfe8475a75d..22c29b187daf 100644
--- a/net/tipc/name_distr.c
+++ b/net/tipc/name_distr.c
@@ -299,7 +299,7 @@ static bool tipc_update_nametbl(struct net *net, struct distr_item *i,
 	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;
diff --git a/net/tipc/name_table.c b/net/tipc/name_table.c
index 962b82956032..0b03d407242f 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;
@@ -341,8 +341,11 @@ static bool tipc_service_insert_publ(struct net *net,
 
 	spin_lock_bh(&sc->lock);
 	sr = tipc_service_create_range(sc, p);
-	if (!sr)
-		goto  exit;
+	if (!sr) {
+		if (err)
+			*err = -ENOBUFS;
+		goto exit;
+	}
 
 	first = list_empty(&sr->all_publ);
 
@@ -353,6 +356,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;
 		}
 	}
@@ -478,20 +483,30 @@ 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) {
+		if (tipc_service_insert_publ(net, sc, p, err))
+			return p;
+	} else if (err) {
+		*err = -ENOBUFS;
+	}
+
 	kfree(p);
 	return NULL;
 }
@@ -760,7 +775,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 +784,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 +805,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 +815,14 @@ 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).
+		 * For cluster-cope one, skb is NULL when named_prepare_buf()
+		 * returns NULL in tipc_named_publish().
+		 */
+		if (!skb && p->scope == TIPC_CLUSTER_SCOPE && err)
+			*err = -ENOBUFS;
+	} else if ((error == -ENOBUFS) && err) {
+		*err = -ENOBUFS;
 	}
 	rc_dests = nt->rc_dests;
 exit:
diff --git a/net/tipc/name_table.h b/net/tipc/name_table.h
index 7ff6eeebaae6..3a15dcfcc2cd 100644
--- a/net/tipc/name_table.h
+++ b/net/tipc/name_table.h
@@ -126,13 +126,14 @@ 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);
 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..c4b274320c9c 100644
--- a/net/tipc/net.c
+++ b/net/tipc/net.c
@@ -138,7 +138,7 @@ 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);
+	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 d84b0274fefc..d99049e67033 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 err = 0;
 
 	if (likely(!flags)) {
 		write_unlock_bh(&n->lock);
@@ -429,12 +430,15 @@ 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);
 	}
+
+	if (err == -ENOBUFS)
+		tipc_node_link_down(n, bearer_id, false);
 }
 
 static void tipc_node_assign_peer_net(struct tipc_node *n, u32 hash_mixes)
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

[PATCH net v2 5/6] 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 clones a list of messages in sequence or delivers the list to
  the peer namespace. When tipc_named_node_up() returns -ENOBUFS,
  tipc_node_link_down() is called from tipc_node_write_unlock()
  instead.

Fixes: 988f3f1603d4 ("tipc: eliminate message disordering during binding table update")
Signed-off-by: Tung Nguyen <tung.quang.nguyen@est.tech>
---
v2: Correct the bearer id that needs to be reset.
v1: https://lore.kernel.org/netdev/20261028033722.125477-2-tung.quang.nguyen@est.tech/

 net/tipc/name_distr.c |  32 +++++----
 net/tipc/name_distr.h |   3 +-
 net/tipc/name_table.c |   1 -
 net/tipc/name_table.h |   2 -
 net/tipc/node.c       | 149 ++++++++++++++++++++++++++++++++----------
 net/tipc/node.h       |   2 +
 6 files changed, 140 insertions(+), 49 deletions(-)

diff --git a/net/tipc/name_distr.c b/net/tipc/name_distr.c
index 22c29b187daf..82684bc7399a 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
+ * @bearer_id: id of the bearer
+ *
+ * Return:
+ * * 0               - Success
+ * * -ENOBUFS        - No buffer space is available
+ * * -EHOSTUNREACH   - Destination node does not exist or cannot be reached
+ * * -EMSGSIZE       - Message's size is too big
+ *
  */
-void tipc_named_node_up(struct net *net, u32 dnode, u16 capabilities)
+int tipc_named_node_up(struct net *net, u32 dnode,
+		       u16 capabilities, int *bearer_id)
 {
 	struct name_table *nt = tipc_name_table(net);
 	struct tipc_net *tn = tipc_net(net);
 	struct sk_buff_head head;
-	u16 seqno;
+	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;
+
+	return rc;
 }
 
 /**
diff --git a/net/tipc/name_distr.h b/net/tipc/name_distr.h
index c677f6f082df..767f9f69624d 100644
--- a/net/tipc/name_distr.h
+++ b/net/tipc/name_distr.h
@@ -69,7 +69,8 @@ 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, int *bearer_id);
 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 0b03d407242f..739450204e5e 100644
--- a/net/tipc/name_table.c
+++ b/net/tipc/name_table.c
@@ -946,7 +946,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);
 	nt->rc_dests = 0;
 	tn->nametbl = nt;
 	spin_lock_init(&tn->nametbl_lock);
diff --git a/net/tipc/name_table.h b/net/tipc/name_table.h
index 3a15dcfcc2cd..a66975ef8546 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 d99049e67033..95f4284cc434 100644
--- a/net/tipc/node.c
+++ b/net/tipc/node.c
@@ -400,6 +400,7 @@ static void tipc_node_write_unlock(struct tipc_node *n)
 	struct net *net = n->net;
 	u32 flags = n->action_flags;
 	struct list_head *publ_list;
+	int rc = 0, bearer = 0;
 	struct tipc_uaddr ua;
 	u32 bearer_id, node;
 	int err = 0;
@@ -426,7 +427,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, &bearer);
 
 	if (flags & TIPC_NOTIFY_LINK_UP) {
 		tipc_mon_peer_up(net, node, bearer_id);
@@ -437,8 +438,15 @@ static void tipc_node_write_unlock(struct tipc_node *n)
 		tipc_nametbl_withdraw(net, &ua, &sk, sk.ref);
 	}
 
-	if (err == -ENOBUFS)
+	if (rc == -ENOBUFS)
+		tipc_node_link_down(n, bearer, false);
+
+	if (err == -ENOBUFS) {
+		if (rc == -ENOBUFS && bearer == bearer_id)
+			return;
+
 		tipc_node_link_down(n, bearer_id, false);
+	}
 }
 
 static void tipc_node_assign_peer_net(struct tipc_node *n, u32 hash_mixes)
@@ -1686,6 +1694,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
@@ -1693,17 +1746,16 @@ static void tipc_lxc_xmit(struct net *peer_net, struct sk_buff_head *list)
  * @dnode: address of destination node
  * @selector: a number used for deterministic link selection
  * Consumes the buffer chain.
- * Return: 0 if success, otherwise: -ELINKCONG,-EHOSTUNREACH,-EMSGSIZE,-ENOBUF
+ * Return: 0 if success, otherwise: -ELINKCONG,-EHOSTUNREACH,-EMSGSIZE,-ENOBUFS
  */
 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)) {
@@ -1719,44 +1771,73 @@ 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)) {
+		if (!skb_queue_empty(&xmitq))
+			__skb_queue_purge(&xmitq);
+		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
+ * * -ENOBUFS       - 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)) {
 		if (!skb_queue_empty(&xmitq))
 			__skb_queue_purge(&xmitq);
-		tipc_node_link_down(n, bearer_id, false);
 	} else {
-		tipc_bearer_xmit(net, bearer_id, &xmitq, &le->maddr, n);
+		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 v2 6/6] 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()'
(Thread 2), 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.

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>
---
v2: Set node->dist_bulk to true when the node comes up, remove an unnecessary
    publication insertion in tipc_update_nametbl().
v1: https://lore.kernel.org/netdev/20261028033722.125477-3-tung.quang.nguyen@est.tech/

 net/tipc/core.c       |  1 +
 net/tipc/core.h       |  2 +
 net/tipc/name_distr.c | 93 ++++++++++++++++++++++++++++++++++++++++---
 net/tipc/name_distr.h |  1 +
 net/tipc/name_table.c | 86 +++++++++++++++++++++++++++++++++------
 net/tipc/name_table.h |  4 ++
 net/tipc/net.c        | 19 ++++++++-
 net/tipc/node.c       | 52 ++++++++++++++++++++----
 net/tipc/node.h       |  1 +
 9 files changed, 231 insertions(+), 28 deletions(-)

diff --git a/net/tipc/core.c b/net/tipc/core.c
index 315975c3be81..351ced173866 100644
--- a/net/tipc/core.c
+++ b/net/tipc/core.c
@@ -61,6 +61,7 @@ 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_self_node_exist = false;
 	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..8f5a94dbc1fc 100644
--- a/net/tipc/core.h
+++ b/net/tipc/core.h
@@ -145,6 +145,8 @@ struct tipc_net {
 	struct work_struct work;
 	/* The numbers of work queues in schedule */
 	atomic_t wq_count;
+	/* 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 82684bc7399a..cc0c17a67863 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,82 @@ 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);
+
+	/* Node's self-address is not set yet */
+	if (!tipc_own_addr(net)) {
+		/* Defer the distribution to node's timer */
+		tipc_node_set_dist_bulk(net, dnode);
+		return 0;
+	}
+
+	/* Node's self-address publication does not exist */
+	if (!tn->nt_self_node_exist) {
+		if (tipc_nametb_insert_self_node_pub(net))
+			return -ENOBUFS;
+
+		tn->nt_self_node_exist = true;
+	}
+
+	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
+ * * -EMSGSIZE       - Message's size is too big
+ *
+ */
+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;
+
+	return rc;
 }
 
 /**
@@ -219,8 +301,9 @@ int tipc_named_node_up(struct net *net, u32 dnode,
 	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);
 
diff --git a/net/tipc/name_distr.h b/net/tipc/name_distr.h
index 767f9f69624d..6bec8b2100c1 100644
--- a/net/tipc/name_distr.h
+++ b/net/tipc/name_distr.h
@@ -76,5 +76,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 739450204e5e..206fcbef8d61 100644
--- a/net/tipc/name_table.c
+++ b/net/tipc/name_table.c
@@ -374,8 +374,8 @@ static bool tipc_service_insert_publ(struct net *net,
 	res = true;
 exit:
 	if (!res)
-		pr_warn("Failed to bind to %u,%u,%u\n",
-			p->sr.type, p->sr.lower, p->sr.upper);
+		pr_warn_ratelimited("Failed to bind to %u,%u,%u\n",
+				    p->sr.type, p->sr.lower, p->sr.upper);
 	spin_unlock_bh(&sc->lock);
 	return res;
 }
@@ -772,26 +772,34 @@ void tipc_nametbl_build_group(struct net *net, struct tipc_group *grp,
 	rcu_read_unlock();
 }
 
-/* tipc_nametbl_publish - add service binding to name table
+/**
+ * tipc_nametbl_insert - insert service binding into name table
+ * @net: network namespace
+ * @ua: the service range is binding to
+ * @sk: address publishing socket
+ * @key: target publication key
+ * @skb: buffer containing the publication
+ * @err: error code
+ *
+ * Return:
+ * * 0          - Success
+ * * -ENOBUFS   - No buffer space is available
+ *
  */
-struct publication *tipc_nametbl_publish(struct net *net, struct tipc_uaddr *ua,
-					 struct tipc_socket_addr *sk,
-					 u32 key, int *err)
+struct publication *tipc_nametbl_insert(struct net *net, struct tipc_uaddr *ua,
+					struct tipc_socket_addr *sk, u32 key,
+					struct sk_buff **skb, int *err)
 {
 	struct name_table *nt = tipc_name_table(net);
 	u32 max_user_pub = TIPC_MAX_PUBL - 1;
-	struct tipc_net *tn = tipc_net(net);
 	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 ||
 	    ua->sr.type == TIPC_TOP_SRV)
 		protocol_type = true;
 
-	spin_lock_bh(&tn->nametbl_lock);
 	if (protocol_type)
 		goto insert;
 
@@ -814,24 +822,76 @@ 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);
+		*skb = tipc_named_publish(net, p);
 		/* Local-scope publication is not published (skb is NULL).
 		 * For cluster-cope one, skb is NULL when named_prepare_buf()
 		 * returns NULL in tipc_named_publish().
 		 */
-		if (!skb && p->scope == TIPC_CLUSTER_SCOPE && err)
+		if (!*skb && p->scope == TIPC_CLUSTER_SCOPE && err)
 			*err = -ENOBUFS;
 	} else if ((error == -ENOBUFS) && err) {
 		*err = -ENOBUFS;
 	}
-	rc_dests = nt->rc_dests;
 exit:
+	return p;
+}
+
+/* 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, int *err)
+{
+	struct name_table *nt = tipc_name_table(net);
+	struct tipc_net *tn = tipc_net(net);
+	struct publication *p = NULL;
+	struct sk_buff *skb = NULL;
+	u32 rc_dests;
+
+	spin_lock_bh(&tn->nametbl_lock);
+	p = tipc_nametbl_insert(net, ua, sk, key, &skb, err);
+	rc_dests = nt->rc_dests;
 	spin_unlock_bh(&tn->nametbl_lock);
 
 	if (skb)
 		tipc_node_broadcast(net, skb, rc_dests);
 	return p;
+}
+
+/**
+ * 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 publication *p = NULL;
+	struct tipc_socket_addr sk;
+	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);
+
+	p = tipc_nametbl_insert_publ(net, &ua, &sk, tn->node_addr, &err);
+
+	if (p) {
+		list_add_tail(&p->binding_node, &nt->cluster_scope);
+		return 0;
+
+	} else if (err == -EEXIST) {
+		return 0;
+	}
 
+	return -ENOBUFS;
 }
 
 /**
diff --git a/net/tipc/name_table.h b/net/tipc/name_table.h
index a66975ef8546..4d421548705c 100644
--- a/net/tipc/name_table.h
+++ b/net/tipc/name_table.h
@@ -123,6 +123,9 @@ bool tipc_nametbl_lookup_group(struct net *net, struct tipc_uaddr *ua,
 			       u32 exclude, bool mcast);
 void tipc_nametbl_build_group(struct net *net, struct tipc_group *grp,
 			      struct tipc_uaddr *ua);
+struct publication *tipc_nametbl_insert(struct net *net, struct tipc_uaddr *ua,
+					struct tipc_socket_addr *sk, u32 key,
+					struct sk_buff **skb, int *err);
 struct publication *tipc_nametbl_publish(struct net *net, struct tipc_uaddr *ua,
 					 struct tipc_socket_addr *sk,
 					 u32 key, int *err);
@@ -132,6 +135,7 @@ struct publication *tipc_nametbl_insert_publ(struct net *net,
 					     struct tipc_uaddr *ua,
 					     struct tipc_socket_addr *sk,
 					     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 c4b274320c9c..c15d8ecf784d 100644
--- a/net/tipc/net.c
+++ b/net/tipc/net.c
@@ -125,9 +125,14 @@ int tipc_net_init(struct net *net, u8 *node_id, u32 addr)
 
 static void tipc_net_finalize(struct net *net, u32 addr)
 {
-	struct tipc_net *tn = tipc_net(net);
+	struct name_table *nt = tipc_name_table(net);
 	struct tipc_socket_addr sk = {0, addr};
+	struct tipc_net *tn = tipc_net(net);
+	struct sk_buff *skb = NULL;
+	struct publication *p;
 	struct tipc_uaddr ua;
+	u32 rc_dests;
+	int err = 0;
 
 	tipc_uaddr(&ua, TIPC_SERVICE_RANGE, TIPC_CLUSTER_SCOPE,
 		   TIPC_NODE_STATE, addr, addr);
@@ -138,7 +143,17 @@ 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, NULL);
+	spin_lock_bh(&tn->nametbl_lock);
+	if (!tn->nt_self_node_exist)
+		p = tipc_nametbl_insert(net, &ua, &sk, addr, &skb, &err);
+
+	if (p || err == -EEXIST)
+		tn->nt_self_node_exist = true;
+	rc_dests = nt->rc_dests;
+	spin_unlock_bh(&tn->nametbl_lock);
+
+	if (skb)
+		tipc_node_broadcast(net, skb, rc_dests);
 }
 
 void tipc_net_finalize_work(struct work_struct *work)
diff --git a/net/tipc/node.c b/net/tipc/node.c
index 95f4284cc434..50d26f537599 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:
@@ -421,6 +423,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)
@@ -438,8 +442,12 @@ static void tipc_node_write_unlock(struct tipc_node *n)
 		tipc_nametbl_withdraw(net, &ua, &sk, sk.ref);
 	}
 
-	if (rc == -ENOBUFS)
-		tipc_node_link_down(n, bearer, false);
+	if (rc == -ENOBUFS) {
+		/* Force the node down to retry distribution */
+		for (bearer = 0; bearer < MAX_BEARERS; bearer++)
+			tipc_node_link_down(n, bearer, false);
+		return;
+	}
 
 	if (err == -ENOBUFS) {
 		if (rc == -ENOBUFS && bearer == bearer_id)
@@ -449,6 +457,20 @@ static void tipc_node_write_unlock(struct tipc_node *n)
 	}
 }
 
+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);
+		if (node_is_up(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);
@@ -576,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);
@@ -815,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)) {
@@ -832,13 +857,22 @@ 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) {
@@ -847,11 +881,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
-- 
2.53.0

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

[kernel test robot]

Hi Tung,

kernel test robot noticed the following build warnings:

[auto build test WARNING on net/main]

url:    https://github.com/intel-lab-lkp/linux/commits/Tung-Nguyen/tipc-fix-memory-leak-in-tipc_node_xmit/20261102-104027
base:   net/main
patch link:    https://lore.kernel.org/r/20261102034207.10639-7-tung.quang.nguyen%40est.tech
patch subject: [PATCH net v2 6/6] tipc: fix NULL pointer dereference in named_distribute()
config: sparc64-randconfig-1001-20261102 (https://download.01.org/0day-ci/archive/20261102/202611022356.UddfVq4D-lkp@intel.com/config)
compiler: clang version 20.1.8 (https://github.com/llvm/llvm-project 87f0227cb60147a26a1eeb4fb06e3b505e9c7261)
reproduce (this is a W=1 build): (https://download.01.org/0day-ci/archive/20261102/202611022356.UddfVq4D-lkp@intel.com/reproduce)

If you fix the issue in a separate patch/commit (i.e. not just a new version of
the same patch/commit), kindly add following tags
| Reported-by: kernel test robot <lkp@intel.com>
| Closes: https://lore.kernel.org/oe-kbuild-all/202611022356.UddfVq4D-lkp@intel.com/

All warnings (new ones prefixed by >>):

>> net/tipc/net.c:147:6: warning: variable 'p' is used uninitialized whenever 'if' condition is false [-Wsometimes-uninitialized]
     147 |         if (!tn->nt_self_node_exist)
         |             ^~~~~~~~~~~~~~~~~~~~~~~
   net/tipc/net.c:150:6: note: uninitialized use occurs here
     150 |         if (p || err == -EEXIST)
         |             ^
   net/tipc/net.c:147:2: note: remove the 'if' if its condition is always true
     147 |         if (!tn->nt_self_node_exist)
         |         ^~~~~~~~~~~~~~~~~~~~~~~~~~~~
     148 |                 p = tipc_nametbl_insert(net, &ua, &sk, addr, &skb, &err);
   net/tipc/net.c:132:23: note: initialize the variable 'p' to silence this warning
     132 |         struct publication *p;
         |                              ^
         |                               = NULL
   1 warning generated.


vim +147 net/tipc/net.c

   125	
   126	static void tipc_net_finalize(struct net *net, u32 addr)
   127	{
   128		struct name_table *nt = tipc_name_table(net);
   129		struct tipc_socket_addr sk = {0, addr};
   130		struct tipc_net *tn = tipc_net(net);
   131		struct sk_buff *skb = NULL;
   132		struct publication *p;
   133		struct tipc_uaddr ua;
   134		u32 rc_dests;
   135		int err = 0;
   136	
   137		tipc_uaddr(&ua, TIPC_SERVICE_RANGE, TIPC_CLUSTER_SCOPE,
   138			   TIPC_NODE_STATE, addr, addr);
   139	
   140		if (cmpxchg(&tn->node_addr, 0, addr))
   141			return;
   142		tipc_set_node_addr(net, addr);
   143		tipc_named_reinit(net);
   144		tipc_sk_reinit(net);
   145		tipc_mon_reinit_self(net);
   146		spin_lock_bh(&tn->nametbl_lock);
 > 147		if (!tn->nt_self_node_exist)
   148			p = tipc_nametbl_insert(net, &ua, &sk, addr, &skb, &err);
   149	
   150		if (p || err == -EEXIST)
   151			tn->nt_self_node_exist = true;
   152		rc_dests = nt->rc_dests;
   153		spin_unlock_bh(&tn->nametbl_lock);
   154	
   155		if (skb)
   156			tipc_node_broadcast(net, skb, rc_dests);
   157	}
   158	

--
0-DAY CI Kernel Test Service
https://github.com/intel/lkp-tests/wiki