Fix AMQP wildcard

Signed-off-by: dusan <borovcanindusan1@gmail.com>

fluxmq/api/http/hooks.go

@@ -83,6 +83,11 @@ func handleHook(ctx context.Context, parser messaging.TopicParser, req hookReque
 }
 
 func resolveHookTopic(ctx context.Context, parser messaging.TopicParser, req hookRequest) (string, error) {
+	hook := strings.ToLower(strings.TrimSpace(req.Hook))
+	if isAMQP091MessageStreamConsume(req, hook) {
+		return strings.TrimPrefix(strings.TrimSpace(req.Topic), "/"), nil
+	}
+
 	if !isMessageTopic(req.Topic) {
 		return "", nil
 	}
@@ -94,7 +99,7 @@ func resolveHookTopic(ctx context.Context, parser messaging.TopicParser, req hoo
 	var topicType messaging.TopicType
 	var err error
 
-	switch strings.ToLower(strings.TrimSpace(req.Hook)) {
+	switch hook {
 	case hookAuthOnPublish:
 		domainID, channelID, subtopic, topicType, err = parser.ParsePublishTopic(ctx, req.Topic, true)
 	case hookAuthOnSubscribe, hookAuthOnUnsubscribe:
@@ -117,3 +122,24 @@ func isMessageTopic(topic string) bool {
 	topic = strings.TrimPrefix(topic, "/")
 	return strings.HasPrefix(topic, string(messaging.MsgTopicPrefix)+"/")
 }
+
+// isAMQP091MessageStreamConsume reports whether the request is an AMQP 0-9-1
+// stream-queue consume of the full message firehose (m/#), which is passed
+// through without parsing because the topic parser cannot resolve a
+// channel-level wildcard.
+//
+// SECURITY: in the default deployment the auth callout is disabled for
+// amqp091 (docker/fluxmq/node*.yaml), so this allow is the only gate for
+// stream consume. The amqp091 listener must remain network-restricted until
+// identity-gated authorization for m/# lands in the gRPC Authorize path.
+func isAMQP091MessageStreamConsume(req hookRequest, hook string) bool {
+	if hook != hookAuthOnSubscribe && hook != hookAuthOnUnsubscribe {
+		return false
+	}
+	if strings.ToLower(strings.TrimSpace(req.Protocol)) != "amqp091" {
+		return false
+	}
+
+	topic := strings.TrimPrefix(strings.TrimSpace(req.Topic), "/")
+	return topic == string(messaging.MsgTopicPrefix)+"/#"
+}

fluxmq/api/http/hooks_test.go

@@ -84,6 +84,52 @@ func TestHooksHandlerUsesSubscribeParserForSubscribeAndUnsubscribe(t *testing.T)
 	}
 }
 
+func TestHooksHandlerAllowsAMQP091MessageStreamWildcard(t *testing.T) {
+	cases := []struct {
+		desc     string
+		protocol string
+		topic    string
+	}{
+		{desc: "plain topic", protocol: "amqp091", topic: "m/#"},
+		{desc: "leading slash topic", protocol: "amqp091", topic: "/m/#"},
+		{desc: "uppercase protocol", protocol: "AMQP091", topic: "m/#"},
+	}
+	for _, tc := range cases {
+		for _, hook := range []string{hookAuthOnSubscribe, hookAuthOnUnsubscribe} {
+			parser := &fakeHookParser{err: errors.New("must not parse stream queue wildcard")}
+			req := httptest.NewRequest("POST", "/hooks", strings.NewReader(`{"hook":"`+hook+`","protocol":"`+tc.protocol+`","topic":"`+tc.topic+`"}`))
+			w := httptest.NewRecorder()
+
+			MakeHooksHandler(parser).ServeHTTP(w, req)
+
+			require.Equal(t, 200, w.Code, tc.desc)
+			require.False(t, parser.publishCalled, tc.desc)
+			require.False(t, parser.subscribeCalled, tc.desc)
+
+			var res hookResponse
+			require.NoError(t, json.NewDecoder(w.Body).Decode(&res), tc.desc)
+			require.Equal(t, hookResultOK, res.Result, tc.desc)
+			require.Equal(t, "m/#", res.Topic, tc.desc)
+		}
+	}
+}
+
+func TestHooksHandlerStillParsesMQTTMessageWildcard(t *testing.T) {
+	parser := &fakeHookParser{err: errors.New("malformed topic")}
+	req := httptest.NewRequest("POST", "/hooks", strings.NewReader(`{"hook":"auth_on_subscribe","protocol":"mqtt","topic":"m/#"}`))
+	w := httptest.NewRecorder()
+
+	MakeHooksHandler(parser).ServeHTTP(w, req)
+
+	require.Equal(t, 200, w.Code)
+	require.False(t, parser.publishCalled)
+	require.True(t, parser.subscribeCalled)
+
+	var res hookResponse
+	require.NoError(t, json.NewDecoder(w.Body).Decode(&res))
+	require.Equal(t, hookResultDeny, res.Result)
+}
+
 func TestHooksHandlerReturnsOKForNonMGTopic(t *testing.T) {
 	req := httptest.NewRequest("POST", "/hooks", strings.NewReader(`{"hook":"auth_on_publish","topic":"$SYS/broker/uptime"}`))
 	w := httptest.NewRecorder()