clusterer_controller: add zero-config multicast HA controller for clusterer

Introduces the clusterer_controller module: a zero-configuration high
availability coordinator for the clusterer module. Nodes discover each other
over authenticated, encrypted UDP multicast, elect a master deterministically
and drive clusterer sharing-tag failover automatically - no per-node node_id
lists or static topology configuration required.

Module (modules/clusterer_controller):
- UDP multicast discovery and keepalive protocol. Every packet's payload is
  sealed with an AEAD behind a 2-byte cleartext magic (key-tier selector) and a
  2-byte cluster_id, both bound into the tag as AAD (so a captured packet can't
  be re-stamped onto another cluster_id on a shared multicast+password group).
  The cluster_id is also filtered before decryption, so several clusters can
  share one multicast group. Two key tiers: a bootstrap key used for the
  admission handshake and the split-brain beacon, derived from the shared
  password with a memory-hard KDF; and a session key (HKDF over an X25519
  ECDH-agreed master salt) for all normal traffic. A startup warning fires if
  the password is the default or has low estimated entropy.
- Crypto suite selected at build time: AES-256-GCM + scrypt (N=2^16) via
  WolfSSL by default, or XChaCha20-Poly1305 + Argon2id when built against
  libsodium (192-bit nonce; detected via pkg-config, linked dynamically). The
  two wire formats are not interoperable, so all nodes must be built alike; the
  active suite is logged at startup.
- Per-worker X25519 ECDH key agreement. The session key is generated once, when
  the first node bootstraps the cluster, and preserved across every master
  change - a new master reuses the key every member already holds, so
  transitions need no re-keying and no re-JOIN cycle.
- Master / backup / member roles with a master_stickiness modparam (default 1):
  a live master is not preempted by a higher-IP node that joins (it becomes the
  backup instead); on master failure the backup - the highest-IP survivor - is
  promoted immediately. With master_stickiness=0 the highest-IP node always
  becomes master.
- Split-brain is prevented and healed by three mechanisms: (1) prevention at
  join time - simultaneously-starting nodes see each other's JOIN_REQs, so a
  node that has seen a higher-IP starter defers self-promotion and joins it
  instead of forming an independent-key lone master; (2) same-key yield - two
  masters sharing a session key see each other's MASTER_ALIVE and the lower-IP
  one yields; (3) divergent-key merge - masters with different session keys
  emit a bootstrap-key MASTER_BEACON, and a node hearing a superior beacon
  (larger partition, ties by higher IP) re-joins that master and adopts its key.
- Fast failure detection (MASTER_ALIVE at 1s, 3s timeout) with immediate backup
  promotion; graceful KEY_HANDOFF + GOODBYE on clean shutdown.
- Single event-driven worker on the OpenSIPS reactor (epoll). Per-source rate
  limiting and 32-bit sequence-number replay protection before/after decrypt;
  peer-table exhaustion defence.
- Join authentication: the master sends an encrypted, unforgeable JOIN_REJECT
  after repeated bootstrap-decrypt failures from an IP. A wrong-password node
  cannot read that reject, so it also self-detects: while joining it counts
  undecryptable packets from other peers and, at the join deadline, shuts down
  ("cannot authenticate - wrong password?") instead of self-promoting into a
  lone split-brain master. Undecryptable session packets from anything other
  than the current master are ignored, so a wrong-password or malicious node on
  the group cannot force the cluster into a re-JOIN churn.
- Sharing-tag control: normally the master is the sole active holder. An
  operator can override this with cc_shtag_force (pin the active tag to a chosen
  node) and revert with cc_shtag_auto; the override is carried in MEMBER_LIST,
  survives master fail-over and auto-clears if the forced node departs. Each
  node logs why its tags were (de)activated.
- Per-cluster configuration via the "cluster" modparam, with global defaults
  for password, query_time, manage_shtags and master_stickiness that each
  cluster can override. MI commands: cc_list_members, cc_node_info,
  cc_list_config (resolved per-cluster settings, including shtag_mode),
  cc_shtag_force and cc_shtag_auto.

clusterer integration (modules/clusterer):
- New controller API (clusterer_ctrl.[ch]): dynamic node add/remove, identity
  update and sharing-tag control driven by the controller.
- Controller-managed sharing tags: forced to backup on startup with MI/script
  tag changes blocked while managed; the active tag follows the elected master.
- Fix a NULL current_node dereference in bin_rcv_cl_packets: with a dynamically
  built topology a cluster can receive BIN packets before this node's identity
  is established; such packets are now dropped until it is, instead of crashing.

core / event_route:
- ipc: expose ipc_is_async_dispatch() to detect being inside an IPC RPC job.
- event_route: raise events inline in that case, avoiding a redundant async
  dispatch.

test/cc_join_reject_test.py: a standalone rogue-joiner security test. From a
non-member host on the multicast segment it sends unauthenticated JOIN_REQs
(expecting a JOIN_REJECT from the master) and a fake-MASTER_ALIVE flood
(expecting the cluster to ignore it), reporting PASS/FAIL. Requires no node
config changes.

Cryptography uses WolfSSL (linked from the tls_wolfssl module) and, optionally,
libsodium for the XChaCha20-Poly1305 + Argon2id build.

Author: Yury Kirsanov

ipc.c

@@ -274,6 +274,7 @@ int ipc_recv_sync_reply(void **param)
 	return 0;
 }
 
+int ipc_running_rpc_job;
 void ipc_handle_job(int fd)
 {
 	ipc_job job;
@@ -301,7 +302,9 @@ void ipc_handle_job(int fd)
 
 	/* custom handling for RPC type */
 	if (job.handler_type==ipc_rpc_type) {
+		ipc_running_rpc_job = 1;
 		((ipc_rpc_f*)job.payload1)( job.snd_proc, job.payload2);
+		ipc_running_rpc_job = 0;
 	} else {
 		/* generic registered type */
 		ipc_handlers[job.handler_type].func( job.snd_proc, job.payload1);

ipc.h

@@ -120,6 +120,15 @@ void ipc_handle_job(int fd);
 void ipc_handle_all_pending_jobs(int fd);
 
 
+/*
+ * A way of checking whether a process is already running within a
+ * ipc_dispatch_rpc() context, in order to avoid double async dispatching,
+ * which is effectively a waste of resources.
+ */
+extern int ipc_running_rpc_job;
+static inline int ipc_is_async_dispatch(void) { return ipc_running_rpc_job; }
+
+
 /* internal functions */
 int init_ipc(void);
 

modules/clusterer/api.h

@@ -327,6 +327,5 @@ static inline module_dependency_t *get_deps_clusterer(const param_export_t *para
 	return alloc_module_dep(MOD_TYPE_DEFAULT, "clusterer", DEP_ABORT);
 }
 
-
 #endif  /* CLUSTERER_API_H */
 

modules/clusterer/clusterer.c

@@ -84,6 +84,7 @@ void sync_check_timer(utime_t ticks, void *param)
 	lock_start_read(cl_list_lock);
 
 	for (cl = *cluster_list; cl; cl = cl->next) {
+		if (!cl->current_node) continue;
 		lock_get(cl->current_node->lock);
 		if (!(cl->current_node->flags & NODE_STATE_ENABLED)) {
 			lock_release(cl->current_node->lock);
@@ -106,11 +107,21 @@ void sync_check_timer(utime_t ticks, void *param)
 						cap->flags &= ~(CAP_SYNC_PENDING|CAP_SYNC_STARTUP);
 						sr_set_status(cl_srg, STR2CI(cap->reg.sr_id), CAP_SR_SYNCED,
 							STR2CI(CAP_SR_STATUS_STR(CAP_SR_SYNCED)), 0);
-						sr_add_report_fmt(cl_srg, STR2CI(cap->reg.sr_id), 0,
-							"ERROR: Sync request aborted! (no donor found in due time)"
-							" => fallback to synced state");
-						LM_ERR("Sync request aborted! (no donor found in due time)"
-						    ", falling back to synced state\n");
+						if (cl->node_list == NULL) {
+							/* no peers — first/lone node, self-sync is expected */
+							sr_add_report_fmt(cl_srg, STR2CI(cap->reg.sr_id), 0,
+								"No peers present — self-synced as first node in cluster");
+							LM_DBG("No peers in cluster %d, capability '%.*s' "
+							    "self-marked as synced (first/lone node)\n",
+							    cl->cluster_id,
+							    cap->reg.name.len, cap->reg.name.s);
+						} else {
+							sr_add_report_fmt(cl_srg, STR2CI(cap->reg.sr_id), 0,
+								"ERROR: Sync request aborted! (no donor found in due time)"
+								" => fallback to synced state");
+							LM_ERR("Sync request aborted! (no donor found in due time)"
+							    ", falling back to synced state\n");
+						}
 						/* send update about the state of this capability */
 						send_single_cap_update(cl, cap, 1);
 
@@ -194,6 +205,9 @@ int cl_set_state(int cluster_id, int node_id, enum cl_node_state state)
 		return 0;
 	}
 
+	if (!cluster->current_node)
+		return -1;
+
 	lock_get(cluster->current_node->lock);
 
 	if (state == STATE_DISABLED && cluster->current_node->flags & NODE_STATE_ENABLED)
@@ -528,7 +542,7 @@ int msg_add_trailer(bin_packet_t *packet, int cluster_id, int dst_id)
 {
 	if (bin_push_int(packet, cluster_id) < 0)
 		return -1;
-	if (bin_push_int(packet, current_id) < 0)
+	if (bin_push_int(packet, GET_CURRENT_ID) < 0)
 		return -1;
 	if (bin_push_int(packet, dst_id) < 0)
 		return -1;
@@ -810,7 +824,7 @@ static void handle_cap_update(bin_packet_t *packet, node_info_t *source)
 	for (i = 0; i < nr_nodes; i++) {
 		bin_pop_int(packet, &node_id);
 
-		if (node_id == current_id) {
+		if (node_id == GET_CURRENT_ID) {
 			bin_pop_int(packet, &nr_cap);
 			for (j = 0; j < nr_cap; j++) {
 				bin_pop_str(packet, &cap);
@@ -991,7 +1005,10 @@ static void handle_remove_node(bin_packet_t *packet, cluster_info_t *cl)
 		return;
 	}
 
-	if (target_node == current_id) {
+	if (target_node == GET_CURRENT_ID) {
+		if (!cl->current_node)
+			return;
+
 		lock_get(cl->current_node->lock);
 
 		if (cl->current_node->flags & NODE_STATE_ENABLED) {
@@ -1038,7 +1055,7 @@ void bin_rcv_cl_extra_packets(bin_packet_t *packet, int packet_type,
 	LM_DBG("received clusterer message from: %s:%hu with source id: %d and"
 			" cluster id: %d\n", ip, port, source_id, cluster_id);
 
-	if (source_id == current_id) {
+	if (source_id == GET_CURRENT_ID) {
 		LM_ERR("Received message with bad source - same node id as this instance\n");
 		return;
 	}
@@ -1094,7 +1111,7 @@ void bin_rcv_cl_extra_packets(bin_packet_t *packet, int packet_type,
 	} else
 		lock_release(node->lock);
 
-	if (dest_id != current_id) {
+	if (dest_id != GET_CURRENT_ID) {
 		/* route the message */
 		bin_push_int(packet, cluster_id);
 		bin_push_int(packet, source_id);
@@ -1172,7 +1189,7 @@ void bin_rcv_cl_packets(bin_packet_t *packet, int packet_type,
 	LM_DBG("received clusterer message from: %s:%hu with source id: %d and "
 		"cluster id: %d\n", ip, port, source_id, cl_id);
 
-	if (source_id == current_id) {
+	if (source_id == GET_CURRENT_ID) {
 		LM_ERR("Received message with bad source - same node id as this instance\n");
 		return;
 	}
@@ -1189,6 +1206,17 @@ void bin_rcv_cl_packets(bin_packet_t *packet, int packet_type,
 		goto exit;
 	}
 
+	/* current_node is legitimately NULL while this node's identity is being
+	 * (re)established for a dynamically constructed cluster (clusterer_ctrl
+	 * update_identity: the cluster can already exist and receive BIN packets
+	 * before current_node is assigned).  Drop the packet instead of
+	 * dereferencing NULL. */
+	if (!cl->current_node) {
+		LM_INFO("Received message for cluster [%d] before local identity is "
+			"established, ignoring\n", cl_id);
+		goto exit;
+	}
+
 	lock_get(cl->current_node->lock);
 	if (!(cl->current_node->flags & NODE_STATE_ENABLED)) {
 		lock_release(cl->current_node->lock);
@@ -1317,7 +1345,7 @@ static void bin_rcv_mod_packets(bin_packet_t *packet, int packet_type,
 	LM_DBG("received bin packet from: %s:%hu with source id: %d and cluster id: %d\n",
 			ip, port, source_id, cluster_id);
 
-	if (source_id == current_id) {
+	if (source_id == GET_CURRENT_ID) {
 		LM_ERR("Received message with bad source - same node id as this instance\n");
 		return;
 	}
@@ -1385,7 +1413,7 @@ static void bin_rcv_mod_packets(bin_packet_t *packet, int packet_type,
 	} else
 		lock_release(node->lock);
 
-	if (dest_id != current_id) {
+	if (dest_id != GET_CURRENT_ID) {
 		/* route the message */
 		bin_push_int(packet, cluster_id);
 		bin_push_int(packet, source_id);
@@ -1460,6 +1488,7 @@ int send_single_cap_update(cluster_info_t *cluster, struct local_cap *cap,
 
 	timestamp = time(NULL);
 
+	if (!cluster->current_node) return -1;
 	lock_get(cluster->current_node->lock);
 
 	for (neigh = cluster->current_node->neighbour_list; neigh;
@@ -1478,7 +1507,7 @@ int send_single_cap_update(cluster_info_t *cluster, struct local_cap *cap,
 		return -1;
 	}
 	bin_push_int(&packet, cluster->cluster_id);
-	bin_push_int(&packet, current_id);
+	bin_push_int(&packet, GET_CURRENT_ID);
 
 	bin_push_int(&packet, ++cluster->current_node->cap_seq_no);
 	bin_push_int(&packet, timestamp);
@@ -1487,7 +1516,7 @@ int send_single_cap_update(cluster_info_t *cluster, struct local_cap *cap,
 
 	/* only the current node */
 	bin_push_int(&packet, 1);
-	bin_push_int(&packet, current_id);
+	bin_push_int(&packet, GET_CURRENT_ID);
 
 	/* only a single capability */
 	bin_push_int(&packet, 1);
@@ -1497,7 +1526,7 @@ int send_single_cap_update(cluster_info_t *cluster, struct local_cap *cap,
 	bin_push_int(&packet, 0);  /* don't require reply */
 
 	bin_push_int(&packet, 1);	/* path length is 1, only current node at this point */
-	bin_push_int(&packet, current_id);
+	bin_push_int(&packet, GET_CURRENT_ID);
 	bin_get_buffer(&packet, &bin_buffer);
 
 	for (i = 0; i < no_dests; i++)
@@ -1545,7 +1574,7 @@ int send_cap_update(node_info_t *dest_node, int require_reply)
 		return -1;
 	}
 	bin_push_int(&packet, dest_node->cluster->cluster_id);
-	bin_push_int(&packet, current_id);
+	bin_push_int(&packet, GET_CURRENT_ID);
 
 	lock_get(dest_node->cluster->current_node->lock);
 
@@ -1560,7 +1589,7 @@ int send_cap_update(node_info_t *dest_node, int require_reply)
 	for (cl_cap = dest_node->cluster->capabilities, nr_cap = 0; cl_cap;
 		cl_cap = cl_cap->next, nr_cap++) ;
 	if (nr_cap) {
-		bin_push_int(&packet, current_id);
+		bin_push_int(&packet, GET_CURRENT_ID);
 		bin_push_int(&packet, nr_cap);
 		for (cl_cap=dest_node->cluster->capabilities;cl_cap;cl_cap=cl_cap->next) {
 			bin_push_str(&packet, &cl_cap->reg.name);
@@ -1591,7 +1620,7 @@ int send_cap_update(node_info_t *dest_node, int require_reply)
 	bin_push_int(&packet, require_reply);
 
 	bin_push_int(&packet, 1);	/* path length is 1, only current node at this point */
-	bin_push_int(&packet, current_id);
+	bin_push_int(&packet, GET_CURRENT_ID);
 	bin_get_buffer(&packet, &bin_buffer);
 
 	if (msg_send(dest_node->cluster->send_sock, dest_node->proto, &dest_node->addr,
@@ -1744,9 +1773,24 @@ int cl_register_cap(str *cap, cl_packet_cb_f packet_cb, cl_event_cb_f event_cb,
 
 	cluster = get_cluster_by_id(cluster_id);
 	if (!cluster) {
-		LM_ERR("cluster id %d is not defined in the %s\n", cluster_id,
-		       db_mode ? "DB" : "script");
-		return -1;
+		if (use_controller) {
+			cluster = shm_malloc(sizeof *cluster);
+			if (!cluster) { LM_ERR("no shm\n"); return -1; }
+			memset(cluster, 0, sizeof *cluster);
+			cluster->cluster_id = cluster_id;
+			if ((cluster->lock = lock_alloc()) == NULL || !lock_init(cluster->lock)) {
+				shm_free(cluster); return -1;
+			}
+			if (cl_list_lock) lock_start_write(cl_list_lock);
+			cluster->next = *cluster_list;
+			*cluster_list = cluster;
+			if (cl_list_lock) lock_stop_write(cl_list_lock);
+			LM_INFO("clusterer: auto-created stub for cluster %d\n", cluster_id);
+		} else {
+			LM_ERR("cluster id %d is not defined in the %s\n", cluster_id,
+			       db_mode ? "DB" : "script");
+			return -1;
+		}
 	}
 
 	new_cl_cap = shm_malloc(sizeof *new_cl_cap + cap->len + CAP_SR_ID_PREFIX_LEN);