From 07e9870f34330776654c29529559e41a2ce7c128 Mon Sep 17 00:00:00 2001
From: Arne Welzel <arne.welzel@corelight.com>
Date: Fri, 20 Jun 2025 17:10:10 +0200
Subject: [PATCH] broker: Handle Broker::publish() with non Broker::Event

Closes #4571
---
 src/broker/messaging.bif                      | 19 ++++---
 .../broker.publish-errors/send..stderr        |  2 +
 .../broker.publish-errors/send..stderr        |  2 +
 testing/btest/broker/publish-errors.zeek      | 52 +++++++++++++++++++
 4 files changed, 69 insertions(+), 6 deletions(-)
 create mode 100644 testing/btest/Baseline.zam/broker.publish-errors/send..stderr
 create mode 100644 testing/btest/Baseline/broker.publish-errors/send..stderr
 create mode 100644 testing/btest/broker/publish-errors.zeek

diff --git a/src/broker/messaging.bif b/src/broker/messaging.bif
index 4347616a97..09f1b54aa5 100644
--- a/src/broker/messaging.bif
+++ b/src/broker/messaging.bif
@@ -66,15 +66,22 @@ static bool publish_event_args(ArgsSpan args, const zeek::String* topic,
                                         "Did you mean to use Cluster::publish() instead of Broker::publish()?");
 
 	if ( args[0]->GetType()->Tag() == zeek::TYPE_RECORD )
-		rval = zeek::broker_mgr->PublishEvent(topic->CheckString(),
-		                                      args[0]->AsRecordVal());
-	else
 		{
-		auto ev = zeek::broker_mgr->MakeEvent(args, frame);
-		rval = zeek::broker_mgr->PublishEvent(topic->CheckString(), ev->AsRecordVal());
+		auto* rv = args[0]->AsRecordVal();
+
+		// same_type() should be fast if it's the type pointers are the same.
+		if ( ! zeek::same_type(rv->GetType(), zeek::BifType::Record::Broker::Event) )
+			{
+			zeek::emit_builtin_error(zeek::util::fmt("expected Broker::Event, got %s",
+			                                         zeek::obj_desc_short(rv->GetType()).c_str()));
+			return false;
+			}
+
+		return zeek::broker_mgr->PublishEvent(topic->CheckString(), rv);
 		}
 
-	return rval;
+	auto ev = zeek::broker_mgr->MakeEvent(args, frame);
+	return zeek::broker_mgr->PublishEvent(topic->CheckString(), ev->AsRecordVal());
 	}
 
 static bool is_cluster_pool(zeek::Val* pool)
diff --git a/testing/btest/Baseline.zam/broker.publish-errors/send..stderr b/testing/btest/Baseline.zam/broker.publish-errors/send..stderr
new file mode 100644
index 0000000000..fc28f8270e
--- /dev/null
+++ b/testing/btest/Baseline.zam/broker.publish-errors/send..stderr
@@ -0,0 +1,2 @@
+### BTest baseline data generated by btest-diff. Do not edit. Use "btest -U/-u" to update. Requires BTest >= 0.63.
+error in ../send.zeek, line 18: expected Broker::Event, got Cluster::Event (Broker::publish(/test/topic, <internal>::evt.18))
diff --git a/testing/btest/Baseline/broker.publish-errors/send..stderr b/testing/btest/Baseline/broker.publish-errors/send..stderr
new file mode 100644
index 0000000000..d8c6153481
--- /dev/null
+++ b/testing/btest/Baseline/broker.publish-errors/send..stderr
@@ -0,0 +1,2 @@
+### BTest baseline data generated by btest-diff. Do not edit. Use "btest -U/-u" to update. Requires BTest >= 0.63.
+error in ../send.zeek, line 18: expected Broker::Event, got Cluster::Event (Broker::publish(/test/topic, evt))
diff --git a/testing/btest/broker/publish-errors.zeek b/testing/btest/broker/publish-errors.zeek
new file mode 100644
index 0000000000..ecf7764664
--- /dev/null
+++ b/testing/btest/broker/publish-errors.zeek
@@ -0,0 +1,52 @@
+# @TEST-DOC: Test that calling Broker::publish() with a Cluster::Event instance fails. Regression test for #4571.
+#
+# @TEST-GROUP: broker
+#
+# @TEST-PORT: BROKER_PORT
+#
+# @TEST-EXEC: btest-bg-run recv "zeek -b ../recv.zeek"
+# @TEST-EXEC: btest-bg-run send "zeek -b ../send.zeek"
+#
+# @TEST-EXEC: btest-bg-wait 30
+#
+# @TEST-EXEC: btest-diff send/.stderr
+
+# @TEST-START-FILE send.zeek
+redef exit_only_after_terminate = T;
+
+event my_event(i: count)
+	{
+	# Not supposed to be invoked!
+	exit(1);
+	}
+
+event zeek_init()
+	{
+	Broker::peer("127.0.0.1", to_port(getenv("BROKER_PORT")));
+	}
+
+event Broker::peer_added(endpoint: Broker::EndpointInfo, msg: string)
+	{
+	print "peer added", msg;
+	local evt = Cluster::make_event(my_event, 42);
+	local r = Broker::publish("/test/topic", evt);
+	assert ! r;  # Supposed to fail.
+	exit(0);
+	}
+# @TEST-END-FILE
+
+
+# @TEST-START-FILE recv.zeek
+
+redef exit_only_after_terminate = T;
+
+event zeek_init()
+	{
+	Broker::listen("127.0.0.1", to_port(getenv("BROKER_PORT")));
+	}
+
+event Broker::peer_lost(endpoint: Broker::EndpointInfo, msg: string)
+	{
+	terminate();
+	}
+# @TEST-END-FILE