Configure ignored and unauthenticated message type which are not disconnected

Author: everesio

README.md

@@ -64,6 +64,16 @@ prerequisites
     watch -c 'curl -s localhost:9090/metrics | grep mqtt | egrep -v '^#''
     ```
 
+## Configuration
+### Examples
+
+- Ignore subscribe / unsubscribe requests
+
+```
+mqtt-proxy server  --mqtt.publisher.name=noop --mqtt.handler.ignore-unsupported SUBSCRIBE --mqtt.handler.ignore-unsupported UNSUBSCRIBE
+```
+
+
 ## Metrics
 
 

cmd/server.go

@@ -42,7 +42,8 @@ func registerServer(m map[string]setupFunc, app *kingpin.Application) {
 	cmd.Flag("mqtt.server-tls-key", "TLS Key for the MQTT server, leave blank to disable TLS").Default("").StringVar(&cfg.MQTT.TLSSrv.Key)
 	cmd.Flag("mqtt.server-tls-client-ca", "TLS CA to verify clients against. If no client CA is specified, there is no client verification on server side. (tls.NoClientCert)").Default("").StringVar(&cfg.MQTT.TLSSrv.ClientCA)
 
-	cmd.Flag("mqtt.handler.allow-unauthenticated", "Allow unauthenticated MQTT requests.").Default("false").BoolVar(&cfg.MQTT.Handler.AllowUnauthenticated)
+	cmd.Flag("mqtt.handler.ignore-unsupported", "List of unsupported messages which are ignored. One of: [SUBSCRIBE, UNSUBSCRIBE]").PlaceHolder("MSG").EnumsVar(&cfg.MQTT.Handler.IgnoreUnsupported, "SUBSCRIBE", "UNSUBSCRIBE")
+	cmd.Flag("mqtt.handler.allow-unauthenticated", "List of messages for which connection is not disconnected if unauthenticated request is received. One of: [PUBLISH, PUBREL, PINGREQ]").PlaceHolder("MSG").EnumsVar(&cfg.MQTT.Handler.AllowUnauthenticated, "PUBLISH", "PUBREL", "PINGREQ")
 	cmd.Flag("mqtt.handler.publish.timeout", "Maximum duration of sending publish request to broker.").Default("0s").DurationVar(&cfg.MQTT.Handler.Publish.Timeout)
 	cmd.Flag("mqtt.handler.publish.async.at-most-once", "Async publish for AT_MOST_ONCE QoS.").Default("false").BoolVar(&cfg.MQTT.Handler.Publish.Async.AtMostOnce)
 	cmd.Flag("mqtt.handler.publish.async.at-least-once", "Async publish for AT_LEAST_ONCE QoS.").Default("false").BoolVar(&cfg.MQTT.Handler.Publish.Async.AtLeastOnce)
@@ -128,6 +129,7 @@ func runServer(
 		}
 
 		handler := mqtthandler.New(logger, registry, publisher,
+			mqtthandler.WithIgnoreUnsupported(cfg.MQTT.Handler.IgnoreUnsupported),
 			mqtthandler.WithAllowUnauthenticated(cfg.MQTT.Handler.AllowUnauthenticated),
 			mqtthandler.WithPublishTimeout(cfg.MQTT.Handler.Publish.Timeout),
 			mqtthandler.WithPublishAsyncAtMostOnce(cfg.MQTT.Handler.Publish.Async.AtMostOnce),

pkg/config/config.go

@@ -35,7 +35,8 @@ type Server struct {
 			ClientCA string
 		}
 		Handler struct {
-			AllowUnauthenticated bool
+			IgnoreUnsupported    []string
+			AllowUnauthenticated []string
 			Publish              struct {
 				Timeout time.Duration
 				Async   struct {

pkg/mqtt/handler/handler.go

@@ -36,22 +36,25 @@ func (h *MQTTHandler) HandleFunc(messageType byte, handlerFunc mqttserver.Handle
 	h.mux.Handle(messageType, handlerFunc)
 }
 
-func (h *MQTTHandler) isAuthenticated(conn mqttserver.Conn, messageName string) bool {
-	if h.opts.allowUnauthenticated {
-		return true
-	}
-	if !conn.Properties().Authenticated() {
-		h.logger.Warnf("Unauthenticated '%s' from /%v", messageName, conn.RemoteAddr())
-		_ = conn.Close()
+func (h *MQTTHandler) disconnectUnauthenticated(conn mqttserver.Conn, packet mqttcodec.ControlPacket) bool {
+	if conn.Properties().Authenticated() {
 		return false
 	}
+	name := packet.Name()
+	for _, v := range h.opts.allowUnauthenticated {
+		if v == name {
+			return false
+		}
+	}
+	h.logger.Warnf("Unauthenticated '%s' from /%v", name, conn.RemoteAddr())
+	_ = conn.Close()
 	return true
 }
 
 func (h *MQTTHandler) handleConnect(conn mqttserver.Conn, packet mqttcodec.ControlPacket) {
 	req := packet.(*mqttcodec.ConnectPacket)
 
-	h.logger.Infof("Handling MQTT message '%s' from /%v", req.MessageName(), conn.RemoteAddr())
+	h.logger.Infof("Handling MQTT message '%s' from /%v", req.Name(), conn.RemoteAddr())
 
 	//TODO: login with user password
 
@@ -72,12 +75,12 @@ func (h *MQTTHandler) handleConnect(conn mqttserver.Conn, packet mqttcodec.Contr
 func (h *MQTTHandler) handlePublish(conn mqttserver.Conn, packet mqttcodec.ControlPacket) {
 	req := packet.(*mqttcodec.PublishPacket)
 
-	h.logger.Debugf("Handling MQTT message '%s' from /%v", req.MessageName(), conn.RemoteAddr())
-
-	if !h.isAuthenticated(conn, req.MessageName()) {
+	if h.disconnectUnauthenticated(conn, packet) {
 		return
 	}
 
+	h.logger.Debugf("Handling MQTT message '%s' from /%v", req.Name(), conn.RemoteAddr())
+
 	var publishCallback apis.PublishCallbackFunc
 
 	switch req.Qos {
@@ -180,11 +183,11 @@ func newPublishRequest(req *mqttcodec.PublishPacket) *apis.PublishRequest {
 func (h *MQTTHandler) handlePublishRelease(conn mqttserver.Conn, packet mqttcodec.ControlPacket) {
 	req := packet.(*mqttcodec.PubrelPacket)
 
-	if !h.isAuthenticated(conn, req.MessageName()) {
+	if h.disconnectUnauthenticated(conn, packet) {
 		return
 	}
 
-	h.logger.Debugf("Handling MQTT message '%s' from /%v", req.MessageName(), conn.RemoteAddr())
+	h.logger.Debugf("Handling MQTT message '%s' from /%v", req.Name(), conn.RemoteAddr())
 	res := mqttcodec.NewControlPacket(mqttcodec.PUBCOMP).(*mqttcodec.PubcompPacket)
 	res.MessageID = req.MessageID
 	err := res.Write(conn)
@@ -196,7 +199,11 @@ func (h *MQTTHandler) handlePublishRelease(conn mqttserver.Conn, packet mqttcode
 func (h *MQTTHandler) handlePing(conn mqttserver.Conn, packet mqttcodec.ControlPacket) {
 	req := packet.(*mqttcodec.PingreqPacket)
 
-	h.logger.Debugf("Handling MQTT message '%s' from /%v", req.MessageName(), conn.RemoteAddr())
+	if h.disconnectUnauthenticated(conn, packet) {
+		return
+	}
+
+	h.logger.Debugf("Handling MQTT message '%s' from /%v", req.Name(), conn.RemoteAddr())
 	res := mqttcodec.NewControlPacket(mqttcodec.PINGRESP)
 	err := res.Write(conn)
 	if err != nil {
@@ -206,13 +213,17 @@ func (h *MQTTHandler) handlePing(conn mqttserver.Conn, packet mqttcodec.ControlP
 
 func (h *MQTTHandler) handleDisconnect(conn mqttserver.Conn, packet mqttcodec.ControlPacket) {
 	req := packet.(*mqttcodec.DisconnectPacket)
-	h.logger.Infof("Handling MQTT message '%s' from /%v", req.MessageName(), conn.RemoteAddr())
+	h.logger.Infof("Handling MQTT message '%s' from /%v", req.Name(), conn.RemoteAddr())
 	err := conn.Close()
 	if err != nil {
 		h.logger.WithError(err).Warnf("Closing connection on 'DISCONNECT' failed")
 	}
 }
 
+func (h *MQTTHandler) ignore(conn mqttserver.Conn, packet mqttcodec.ControlPacket) {
+	h.logger.Debugf("No handler available for MQTT message '%s' from /%v. Ignoring", packet.Name(), conn.RemoteAddr())
+}
+
 func New(logger log.Logger, registry *prometheus.Registry, publisher apis.Publisher, opts ...Option) *MQTTHandler {
 	options := options{}
 	for _, o := range opts {
@@ -230,6 +241,19 @@ func New(logger log.Logger, registry *prometheus.Registry, publisher apis.Publis
 	h.HandleFunc(mqttcodec.DISCONNECT, h.handleDisconnect)
 	h.HandleFunc(mqttcodec.PUBREL, h.handlePublishRelease)
 	h.HandleFunc(mqttcodec.PINGREQ, h.handlePing)
+
+	for _, name := range options.ignoreUnsupported {
+		for t, n := range mqttcodec.MqttMessageTypeNames {
+			if n == name {
+				logger.Infof("%s requests will be ignored", name)
+				h.HandleFunc(t, h.ignore)
+				continue
+			}
+		}
+	}
+	for _, name := range options.allowUnauthenticated {
+		logger.Infof("%s requests will be allow unauthenticated", name)
+	}
 	return h
 
 }

pkg/mqtt/handler/option.go

@@ -3,7 +3,8 @@ package mqtthandler
 import "time"
 
 type options struct {
-	allowUnauthenticated    bool
+	ignoreUnsupported       []string
+	allowUnauthenticated    []string
 	publishTimeout          time.Duration
 	publishAsyncAtMostOnce  bool
 	publishAsyncAtLeastOnce bool
@@ -20,9 +21,15 @@ func (f optionFunc) apply(o *options) {
 	f(o)
 }
 
-func WithAllowUnauthenticated(b bool) Option {
+func WithIgnoreUnsupported(a []string) Option {
 	return optionFunc(func(o *options) {
-		o.allowUnauthenticated = b
+		o.ignoreUnsupported = a
+	})
+}
+
+func WithAllowUnauthenticated(a []string) Option {
+	return optionFunc(func(o *options) {
+		o.allowUnauthenticated = a
 	})
 }
 

pkg/mqtt/server/handler.go

@@ -61,7 +61,8 @@ func (mux *ServeMux) Handle(messageType byte, handler Handler) {
 func (mux *ServeMux) ServeMQTT(c Conn, p mqttcodec.ControlPacket) {
 	entry := mux.m[p.Type()]
 	if entry.h == nil {
-		mux.logger.Debugf("No handler available for MQTT message '%s' from /%v. Ignoring", p.Name(), c.RemoteAddr())
+		mux.logger.Warnf("No handler available for MQTT message '%s' from /%v. Disconnecting", p.Name(), c.RemoteAddr())
+		_ = c.Close()
 		return
 	}
 	entry.h.ServeMQTT(c, p)