Fix AMQP wildcard

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

cmd/fluxmq/main.go

@@ -137,6 +137,7 @@ func main() {
 		connect.WithInterceptors(otelInterceptor),
 	)
 	mux.Handle(path, handler)
+	mux.Handle("/hooks", fluxmqhttp.MakeHooksHandler(parser))
 	mux.HandleFunc("/health", func(w http.ResponseWriter, _ *http.Request) {
 		w.WriteHeader(http.StatusOK)
 		w.Write([]byte(`{"status":"ok"}`)) //nolint:errcheck // HTTP response write; client disconnect is non-fatal.

docker/fluxmq/node1.yaml

@@ -136,3 +136,19 @@ auth:
     coap: true
     amqp: true
     amqp091: false
+
+hooks:
+  url: "http://fluxmq-auth:7016"
+  transport: "http"
+  timeout: 500ms
+  fail_mode: "deny"
+  protocols:
+    mqtt: true
+    http: true
+    coap: true
+    amqp: true
+    amqp091: true
+  events:
+    auth_on_publish: true
+    auth_on_subscribe: true
+    auth_on_unsubscribe: true

docker/fluxmq/node2.yaml

@@ -133,3 +133,19 @@ auth:
     coap: true
     amqp: true
     amqp091: false
+
+hooks:
+  url: "http://fluxmq-auth:7016"
+  transport: "http"
+  timeout: 500ms
+  fail_mode: "deny"
+  protocols:
+    mqtt: true
+    http: true
+    coap: true
+    amqp: true
+    amqp091: true
+  events:
+    auth_on_publish: true
+    auth_on_subscribe: true
+    auth_on_unsubscribe: true

docker/fluxmq/node3.yaml

@@ -133,3 +133,19 @@ auth:
     coap: true
     amqp: true
     amqp091: false
+
+hooks:
+  url: "http://fluxmq-auth:7016"
+  transport: "http"
+  timeout: 500ms
+  fail_mode: "deny"
+  protocols:
+    mqtt: true
+    http: true
+    coap: true
+    amqp: true
+    amqp091: true
+  events:
+    auth_on_publish: true
+    auth_on_subscribe: true
+    auth_on_unsubscribe: true

fluxmq/api/grpc/server.go

@@ -145,9 +145,7 @@ func (s *connectServer) Authorize(ctx context.Context, req *connect.Request[auth
 		return nil, encodeError(err)
 	}
 
-	return connect.NewResponse(&authv1.AuthzRes{
-		Authorized: res.GetAuthorized(),
-	}), nil
+	return connect.NewResponse(&authv1.AuthzRes{Authorized: res.GetAuthorized()}), nil
 }
 
 func shouldTryDomainAuth(msg *authv1.AuthnReq, username, password string) bool {

fluxmq/api/grpc/server_test.go

@@ -0,0 +1,76 @@
+// Copyright (c) Abstract Machines
+// SPDX-License-Identifier: Apache-2.0
+
+package grpc
+
+import (
+	"context"
+	"testing"
+
+	"connectrpc.com/connect"
+	authv1 "github.com/absmach/fluxmq/pkg/proto/auth/v1"
+	"github.com/absmach/magistrala/internal/atom"
+	"github.com/absmach/magistrala/pkg/messaging"
+	"github.com/stretchr/testify/require"
+)
+
+type fakeTopicParser struct {
+	domainID  string
+	channelID string
+	subtopic  string
+	topicType messaging.TopicType
+	err       error
+}
+
+func (p fakeTopicParser) ParsePublishTopic(context.Context, string, bool) (string, string, string, messaging.TopicType, error) {
+	return p.domainID, p.channelID, p.subtopic, p.topicType, p.err
+}
+
+func (p fakeTopicParser) ParseSubscribeTopic(context.Context, string, bool) (string, string, string, messaging.TopicType, error) {
+	return p.domainID, p.channelID, p.subtopic, p.topicType, p.err
+}
+
+type fakeAtomAuthorizer struct {
+	resp atom.AuthzResponse
+	err  error
+}
+
+func (a fakeAtomAuthorizer) CheckAuthz(context.Context, atom.AuthzRequest) (atom.AuthzResponse, error) {
+	return a.resp, a.err
+}
+
+func TestAuthorizeReturnsAuthorizedOnlyWhenAllowed(t *testing.T) {
+	srv := NewServer(nil, nil, fakeTopicParser{
+		domainID:  "26ad5c3f-cd91-4ff0-9685-0c3115643174",
+		channelID: "cdc8f55f-0c54-4a9f-b4aa-8c69d4a8ce15",
+		subtopic:  "messages",
+		topicType: messaging.MessageType,
+	}, fakeAtomAuthorizer{resp: atom.AuthzResponse{Allowed: true}}).(*connectServer)
+
+	res, err := srv.Authorize(context.Background(), connect.NewRequest(&authv1.AuthzReq{
+		ExternalId: "64d6bc95-b313-4412-9369-299543d9c63b",
+		Topic:      "m/d1/c/ch1/messages",
+		Action:     authv1.Action_Publish,
+	}))
+	require.NoError(t, err)
+	require.True(t, res.Msg.GetAuthorized())
+	require.Empty(t, res.Msg.ProtoReflect().GetUnknown())
+}
+
+func TestAuthorizeReturnsDeniedOnlyWhenDenied(t *testing.T) {
+	srv := NewServer(nil, nil, fakeTopicParser{
+		domainID:  "26ad5c3f-cd91-4ff0-9685-0c3115643174",
+		channelID: "cdc8f55f-0c54-4a9f-b4aa-8c69d4a8ce15",
+		subtopic:  "messages",
+		topicType: messaging.MessageType,
+	}, fakeAtomAuthorizer{resp: atom.AuthzResponse{Allowed: false}}).(*connectServer)
+
+	res, err := srv.Authorize(context.Background(), connect.NewRequest(&authv1.AuthzReq{
+		ExternalId: "64d6bc95-b313-4412-9369-299543d9c63b",
+		Topic:      "m/d1/c/ch1/messages",
+		Action:     authv1.Action_Subscribe,
+	}))
+	require.NoError(t, err)
+	require.False(t, res.Msg.GetAuthorized())
+	require.Empty(t, res.Msg.ProtoReflect().GetUnknown())
+}

fluxmq/api/http/hooks.go

@@ -0,0 +1,145 @@
+// Copyright (c) Abstract Machines
+// SPDX-License-Identifier: Apache-2.0
+
+package http
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"net/http"
+	"strings"
+
+	"github.com/absmach/magistrala/pkg/messaging"
+)
+
+const (
+	hookResultOK   = "ok"
+	hookResultDeny = "deny"
+
+	hookAuthOnPublish     = "auth_on_publish"
+	hookAuthOnSubscribe   = "auth_on_subscribe"
+	hookAuthOnUnsubscribe = "auth_on_unsubscribe"
+)
+
+type hookRequest struct {
+	Hook       string            `json:"hook"`
+	ClientID   string            `json:"client_id"`
+	ExternalID string            `json:"external_id"`
+	Protocol   string            `json:"protocol"`
+	Topic      string            `json:"topic"`
+	Payload    []byte            `json:"payload,omitempty"`
+	QoS        uint32            `json:"qos"`
+	Retain     bool              `json:"retain"`
+	Properties map[string]string `json:"properties,omitempty"`
+	Username   string            `json:"username,omitempty"`
+	Password   string            `json:"password,omitempty"`
+}
+
+type hookResponse struct {
+	Result     string            `json:"result"`
+	Topic      string            `json:"topic,omitempty"`
+	Payload    []byte            `json:"payload,omitempty"`
+	PayloadSet bool              `json:"payload_set,omitempty"`
+	QoS        uint32            `json:"qos,omitempty"`
+	QoSSet     bool              `json:"qos_set,omitempty"`
+	Retain     bool              `json:"retain,omitempty"`
+	RetainSet  bool              `json:"retain_set,omitempty"`
+	Properties map[string]string `json:"properties,omitempty"`
+	ExternalID string            `json:"external_id,omitempty"`
+	ReasonCode uint32            `json:"reason_code,omitempty"`
+	Reason     string            `json:"reason,omitempty"`
+}
+
+// MakeHooksHandler returns an HTTP handler for FluxMQ blocking hooks.
+func MakeHooksHandler(parser messaging.TopicParser) http.Handler {
+	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if r.Method != http.MethodPost {
+			writeError(w, http.StatusMethodNotAllowed, "method not allowed")
+			return
+		}
+
+		var req hookRequest
+		if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
+			writeError(w, http.StatusBadRequest, "invalid hook request")
+			return
+		}
+
+		res := handleHook(r.Context(), parser, req)
+		w.Header().Set("Content-Type", contentType)
+		w.WriteHeader(http.StatusOK)
+		if err := json.NewEncoder(w).Encode(res); err != nil {
+			return
+		}
+	})
+}
+
+func handleHook(ctx context.Context, parser messaging.TopicParser, req hookRequest) hookResponse {
+	topic, err := resolveHookTopic(ctx, parser, req)
+	if err != nil {
+		return hookResponse{Result: hookResultDeny, Reason: err.Error()}
+	}
+	return hookResponse{Result: hookResultOK, Topic: topic}
+}
+
+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
+	}
+	if parser == nil {
+		return "", fmt.Errorf("topic parser is not configured")
+	}
+
+	var domainID, channelID, subtopic string
+	var topicType messaging.TopicType
+	var err error
+
+	switch hook {
+	case hookAuthOnPublish:
+		domainID, channelID, subtopic, topicType, err = parser.ParsePublishTopic(ctx, req.Topic, true)
+	case hookAuthOnSubscribe, hookAuthOnUnsubscribe:
+		domainID, channelID, subtopic, topicType, err = parser.ParseSubscribeTopic(ctx, req.Topic, true)
+	default:
+		return "", nil
+	}
+	if err != nil {
+		return "", err
+	}
+	if topicType != messaging.MessageType {
+		return "", nil
+	}
+
+	return messaging.EncodeTopic(domainID, channelID, subtopic), nil
+}
+
+func isMessageTopic(topic string) bool {
+	topic = strings.TrimSpace(topic)
+	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

@@ -0,0 +1,157 @@
+// Copyright (c) Abstract Machines
+// SPDX-License-Identifier: Apache-2.0
+
+package http
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"net/http/httptest"
+	"strings"
+	"testing"
+
+	"github.com/absmach/magistrala/pkg/messaging"
+	"github.com/stretchr/testify/require"
+)
+
+type fakeHookParser struct {
+	domainID        string
+	channelID       string
+	subtopic        string
+	topicType       messaging.TopicType
+	err             error
+	publishCalled   bool
+	subscribeCalled bool
+}
+
+func (p *fakeHookParser) ParsePublishTopic(context.Context, string, bool) (string, string, string, messaging.TopicType, error) {
+	p.publishCalled = true
+	return p.domainID, p.channelID, p.subtopic, p.topicType, p.err
+}
+
+func (p *fakeHookParser) ParseSubscribeTopic(context.Context, string, bool) (string, string, string, messaging.TopicType, error) {
+	p.subscribeCalled = true
+	return p.domainID, p.channelID, p.subtopic, p.topicType, p.err
+}
+
+func TestHooksHandlerReturnsCanonicalTopicModifier(t *testing.T) {
+	parser := &fakeHookParser{
+		domainID:  "26ad5c3f-cd91-4ff0-9685-0c3115643174",
+		channelID: "cdc8f55f-0c54-4a9f-b4aa-8c69d4a8ce15",
+		subtopic:  "messages",
+		topicType: messaging.MessageType,
+	}
+	req := httptest.NewRequest("POST", "/hooks", strings.NewReader(`{
+		"hook":"auth_on_publish",
+		"client_id":"cli1",
+		"external_id":"64d6bc95-b313-4412-9369-299543d9c63b",
+		"protocol":"mqtt",
+		"topic":"m/d1/c/ch1/messages"
+	}`))
+	w := httptest.NewRecorder()
+
+	MakeHooksHandler(parser).ServeHTTP(w, req)
+
+	require.Equal(t, 200, w.Code)
+	require.True(t, parser.publishCalled)
+	require.False(t, parser.subscribeCalled)
+
+	var res hookResponse
+	require.NoError(t, json.NewDecoder(w.Body).Decode(&res))
+	require.Equal(t, hookResultOK, res.Result)
+	require.Equal(t, "m/26ad5c3f-cd91-4ff0-9685-0c3115643174/c/cdc8f55f-0c54-4a9f-b4aa-8c69d4a8ce15/messages", res.Topic)
+}
+
+func TestHooksHandlerUsesSubscribeParserForSubscribeAndUnsubscribe(t *testing.T) {
+	parser := &fakeHookParser{
+		domainID:  "26ad5c3f-cd91-4ff0-9685-0c3115643174",
+		channelID: "cdc8f55f-0c54-4a9f-b4aa-8c69d4a8ce15",
+		subtopic:  "messages/+",
+		topicType: messaging.MessageType,
+	}
+	for _, hook := range []string{hookAuthOnSubscribe, hookAuthOnUnsubscribe} {
+		parser.publishCalled = false
+		parser.subscribeCalled = false
+		req := httptest.NewRequest("POST", "/hooks", strings.NewReader(`{"hook":"`+hook+`","topic":"m/d1/c/ch1/messages/+"}`))
+		w := httptest.NewRecorder()
+
+		MakeHooksHandler(parser).ServeHTTP(w, req)
+
+		require.Equal(t, 200, w.Code)
+		require.False(t, parser.publishCalled)
+		require.True(t, parser.subscribeCalled)
+	}
+}
+
+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()
+
+	MakeHooksHandler(nil).ServeHTTP(w, req)
+
+	require.Equal(t, 200, w.Code)
+	var res hookResponse
+	require.NoError(t, json.NewDecoder(w.Body).Decode(&res))
+	require.Equal(t, hookResultOK, res.Result)
+	require.Empty(t, res.Topic)
+}
+
+func TestHooksHandlerDeniesUnresolvedMGTopic(t *testing.T) {
+	parser := &fakeHookParser{err: errors.New("failed to resolve channel route")}
+	req := httptest.NewRequest("POST", "/hooks", strings.NewReader(`{"hook":"auth_on_publish","topic":"m/d1/c/ch1/messages"}`))
+	w := httptest.NewRecorder()
+
+	MakeHooksHandler(parser).ServeHTTP(w, req)
+
+	require.Equal(t, 200, w.Code)
+	var res hookResponse
+	require.NoError(t, json.NewDecoder(w.Body).Decode(&res))
+	require.Equal(t, hookResultDeny, res.Result)
+}