Add noop and plain authentication

Author: everesio

README.md

@@ -15,10 +15,14 @@ MQTT Proxy allows MQTT clients to send messages to other messaging systems
     * [x] [MQTT 3.1.1](http://docs.oasis-open.org/mqtt/mqtt/v3.1.1/mqtt-v3.1.1.html)
     * [ ] [MQTT 5.0](https://docs.oasis-open.org/mqtt/mqtt/v5.0/mqtt-v5.0.html)
 * Publisher
+    * [x] Noop
     * [x] [Apache Kafka](https://kafka.apache.org/)
     * [ ] [Apache Pulsar](https://pulsar.apache.org/)
     * [ ] Others
-* [ ] Authentication
+* Authentication
+    * [x] Noop
+    * [x] Plain
+    * [ ] Others
 * [x] Helm chart
 
 ## Build
@@ -111,6 +115,38 @@ prerequisites
     mosquitto_pub -m "on" -t "dummy" -k 20 -i mqtt-proxy.clientv --repeat 1 -q 1 -h <ec2-ip> -p 1884
     ```
 
+### plain authenticator
+
+1. start server with `plain` authenticator
+    * with credentials file
+
+    ```
+    cat <<EOF > mqtt-credentials.csv
+    alice,alice-secret
+    "bob","bob-secret"
+    EOF
+    ```
+
+    ```
+    mqtt-proxy server --mqtt.publisher.name=noop \
+        --mqtt.handler.auth.name=plain \
+        --mqtt.handler.auth.plain.credentials-file=mqtt-credentials.csv
+    ```
+    * providing credentials as parameters
+    ```
+    mqtt-proxy server --mqtt.publisher.name=noop \
+        --mqtt.handler.auth.name=plain \
+        --mqtt.handler.auth.plain.credentials=alice=alice-secret \
+        --mqtt.handler.auth.plain.credentials=bob=bob-secret
+    ```
+2. publish
+
+    ```
+    mosquitto_pub -m "on" -t "dummy" -u alice -P alice-secret
+    mosquitto_pub -L mqtt://bob:bob-secret@localhost:1883/dummy -m "on"
+    ```
+
+
 ## Configuration
 
 ### Kafka publisher
@@ -121,7 +157,6 @@ Kafka producer configuration properties used by [librdkafka](https://github.com/
 --mqtt.publisher.kafka.config=producer.sasl.mechanisms=PLAIN,producer.security.protocol=SASL_SSL,producer.sasl.username=myuser,producer.sasl.password=mypasswd
 ```
 
-
 ### Examples
 
 - Ignore subscribe / unsubscribe requests
@@ -141,4 +176,5 @@ metric | labels | description
 |mqtt_proxy_server_connections_total| |Total number of TCP connections from clients to server.|
 |mqtt_proxy_handler_requests_total|type|Total number of MQTT requests labeled by package control type. |
 |mqtt_proxy_handler_responses_total|type|Total number of MQTT responses labeled by package control type. |
-|mqtt_proxy_publisher_publish_duration_seconds | name, type, qos | Histogram tracking latencies for publish requests. |
+|mqtt_proxy_publisher_publish_duration_seconds | name, type, qos | Histogram tracking latencies for publish requests. |
+|mqtt_proxy_authenticator_login_duration_seconds | name, code, err | Histogram tracking latencies for login requests. |

apis/auth.go

@@ -1,6 +1,15 @@
 package apis
 
-import "context"
+import (
+	"context"
+
+	mqttcodec "github.com/grepplabs/mqtt-proxy/pkg/mqtt/codec"
+)
+
+const (
+	AuthAccepted     = mqttcodec.Accepted
+	AuthUnauthorized = mqttcodec.RefusedNotAuthorized
+)
 
 type UserPasswordAuthRequest struct {
 	Username string
@@ -12,10 +21,11 @@ type UserPasswordAuthResponse struct {
 }
 
 type UserPasswordAuthenticator interface {
+	Name() string
 	Login(context.Context, *UserPasswordAuthRequest) (*UserPasswordAuthResponse, error)
 	Close() error
 }
 
 type UserPasswordAuthenticatorFactory interface {
-	New(context.Context) (UserPasswordAuthenticator, error)
+	New(params []string) (UserPasswordAuthenticator, error)
 }

apis/publish.go

@@ -32,5 +32,5 @@ type Publisher interface {
 }
 
 type PublisherFactory interface {
-	New(context.Context) (Publisher, error)
+	New(params []string) (Publisher, error)
 }

cmd/server.go

@@ -2,13 +2,16 @@ package cmd
 
 import (
 	"github.com/grepplabs/mqtt-proxy/apis"
+	authinst "github.com/grepplabs/mqtt-proxy/pkg/auth/instrument"
+	authnoop "github.com/grepplabs/mqtt-proxy/pkg/auth/noop"
+	authplain "github.com/grepplabs/mqtt-proxy/pkg/auth/plain"
 	"github.com/grepplabs/mqtt-proxy/pkg/config"
 	"github.com/grepplabs/mqtt-proxy/pkg/log"
 	mqtthandler "github.com/grepplabs/mqtt-proxy/pkg/mqtt/handler"
 	"github.com/grepplabs/mqtt-proxy/pkg/prober"
-	"github.com/grepplabs/mqtt-proxy/pkg/publisher/instrument"
-	"github.com/grepplabs/mqtt-proxy/pkg/publisher/kafka"
-	"github.com/grepplabs/mqtt-proxy/pkg/publisher/noop"
+	pubinst "github.com/grepplabs/mqtt-proxy/pkg/publisher/instrument"
+	pubkafka "github.com/grepplabs/mqtt-proxy/pkg/publisher/kafka"
+	pubnoop "github.com/grepplabs/mqtt-proxy/pkg/publisher/noop"
 	httpserver "github.com/grepplabs/mqtt-proxy/pkg/server/http"
 	mqttserver "github.com/grepplabs/mqtt-proxy/pkg/server/mqtt"
 	"github.com/grepplabs/mqtt-proxy/pkg/tls"
@@ -25,6 +28,7 @@ func registerServer(m map[string]setupFunc, app *kingpin.Application) {
 	cmd := app.Command(command, "mqtt-proxy server")
 
 	cfg := new(config.Server)
+	cfg.Init()
 
 	cmd.Flag("http.listen-address", "Listen host:port for HTTP endpoints.").Default("0.0.0.0:9090").StringVar(&cfg.HTTP.ListenAddress)
 	cmd.Flag("http.grace-period", "Time to wait after an interrupt received for HTTP Server.").Default("10s").DurationVar(&cfg.HTTP.GracePeriod)
@@ -49,12 +53,17 @@ func registerServer(m map[string]setupFunc, app *kingpin.Application) {
 	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)
 	cmd.Flag("mqtt.handler.publish.async.exactly-once", "Async publish for EXACTLY_ONCE QoS.").Default("false").BoolVar(&cfg.MQTT.Handler.Publish.Async.ExactlyOnce)
 
-	cmd.Flag("mqtt.publisher.name", "Publisher name. One of: [noop, kafka]").Default(config.Noop).EnumVar(&cfg.MQTT.Publisher.Name, config.Noop, config.Kafka)
+	cmd.Flag("mqtt.handler.auth.name", "Authenticator name. One of: [noop, plain]").Default(config.AuthNoop).EnumVar(&cfg.MQTT.Handler.Authenticator.Name, config.AuthNoop, config.AuthPlain)
+	cmd.Flag("mqtt.handler.auth.plain.credentials", "List of username and password fields.").Default("USERNAME=PASSWORD").StringMapVar(&cfg.MQTT.Handler.Authenticator.Plain.Credentials)
+	cmd.Flag("mqtt.handler.auth.plain.credentials-file", "Location of a headerless CSV file containing `usernanme,password` records").Default("").StringVar(&cfg.MQTT.Handler.Authenticator.Plain.CredentialsFile)
+
+	cmd.Flag("mqtt.publisher.name", "Publisher name. One of: [noop, kafka]").Default(config.PublisherNoop).EnumVar(&cfg.MQTT.Publisher.Name, config.PublisherNoop, config.PublisherKafka)
 	cmd.Flag("mqtt.publisher.kafka.config", "Comma separated list of properties").PlaceHolder("PROP=VAL").SetValue(&cfg.MQTT.Publisher.Kafka.ConfArgs)
 	cmd.Flag("mqtt.publisher.kafka.bootstrap-servers", "Kafka bootstrap servers").Default("localhost:9092").StringVar(&cfg.MQTT.Publisher.Kafka.BootstrapServers)
 	cmd.Flag("mqtt.publisher.kafka.grace-period", "Time to wait after an interrupt received for Kafka publisher.").Default("10s").DurationVar(&cfg.MQTT.Publisher.Kafka.GracePeriod)
 	cmd.Flag("mqtt.publisher.kafka.default-topic", "Default Kafka topic for MQTT publish messages").Default("").StringVar(&cfg.MQTT.Publisher.Kafka.DefaultTopic)
 	cmd.Flag("mqtt.publisher.kafka.topic-mappings", "Comma separated list of Kafka topic to MQTT topic mappings").PlaceHolder("TOPIC=REGEX").SetValue(&cfg.MQTT.Publisher.Kafka.TopicMappings)
+	cmd.Flag("mqtt.publisher.kafka.workers", "Number of kafka publisher workers").Default("1").IntVar(&cfg.MQTT.Publisher.Kafka.Workers)
 
 	m[command] = func(group *run.Group, logger log.Logger, registry *prometheus.Registry) error {
 		return runServer(group, logger, registry, cfg)
@@ -91,33 +100,58 @@ func runServer(
 			srv.Shutdown(err)
 		})
 	}
+
+	var authenticator apis.UserPasswordAuthenticator
+	{
+		logger.Infof("setting up authenticator %s", cfg.MQTT.Handler.Authenticator.Name)
+
+		switch cfg.MQTT.Handler.Authenticator.Name {
+		case config.AuthNoop:
+			authenticator = authnoop.New(logger, registry)
+		case config.AuthPlain:
+			authenticator, err = authplain.New(logger, registry,
+				authplain.WithCredentials(cfg.MQTT.Handler.Authenticator.Plain.Credentials),
+				authplain.WithCredentialsFile(cfg.MQTT.Handler.Authenticator.Plain.CredentialsFile),
+			)
+			if err != nil {
+				return errors.Wrap(err, "setup plain authenticator")
+			}
+		default:
+			return errors.Errorf("unknown authenticator %s", cfg.MQTT.Handler.Authenticator.Name)
+		}
+		authenticator = authinst.New(authenticator, registry)
+		defer func() {
+			err := authenticator.Close()
+			if err != nil {
+				logger.WithError(err).Warnf("authenticator close failed")
+			}
+		}()
+	}
 	var publisher apis.Publisher
 	{
-		logger.Infof("setting publisher")
+		logger.Infof("setting up publisher %s", cfg.MQTT.Publisher.Name)
 
 		var err error
 
 		switch cfg.MQTT.Publisher.Name {
-		case config.Noop:
-			publisher, err = noop.New(logger, registry)
-			if err != nil {
-				return errors.Wrap(err, "setup noop publisher")
-			}
-		case config.Kafka:
-			publisher, err = kafka.New(logger, registry,
-				kafka.WithBootstrapServers(cfg.MQTT.Publisher.Kafka.BootstrapServers),
-				kafka.WithDefaultTopic(cfg.MQTT.Publisher.Kafka.DefaultTopic),
-				kafka.WithTopicMappings(cfg.MQTT.Publisher.Kafka.TopicMappings),
-				kafka.WithConfigMap(cfg.MQTT.Publisher.Kafka.ConfArgs.ConfigMap()),
-				kafka.WithGracePeriod(cfg.MQTT.Publisher.Kafka.GracePeriod),
+		case config.PublisherNoop:
+			publisher = pubnoop.New(logger, registry)
+		case config.PublisherKafka:
+			publisher, err = pubkafka.New(logger, registry,
+				pubkafka.WithBootstrapServers(cfg.MQTT.Publisher.Kafka.BootstrapServers),
+				pubkafka.WithDefaultTopic(cfg.MQTT.Publisher.Kafka.DefaultTopic),
+				pubkafka.WithTopicMappings(cfg.MQTT.Publisher.Kafka.TopicMappings),
+				pubkafka.WithConfigMap(cfg.MQTT.Publisher.Kafka.ConfArgs.ConfigMap()),
+				pubkafka.WithGracePeriod(cfg.MQTT.Publisher.Kafka.GracePeriod),
+				pubkafka.WithWorkers(cfg.MQTT.Publisher.Kafka.Workers),
 			)
 			if err != nil {
 				return errors.Wrap(err, "setup kafka publisher")
 			}
 		default:
-			return errors.Errorf("Unknown publisher %s", cfg.MQTT.Publisher.Name)
+			return errors.Errorf("unknown publisher %s", cfg.MQTT.Publisher.Name)
 		}
-		publisher = instrument.New(publisher, registry)
+		publisher = pubinst.New(publisher, registry)
 
 		group.Add(func() error {
 			return publisher.Serve()
@@ -140,6 +174,7 @@ func runServer(
 			mqtthandler.WithPublishAsyncAtMostOnce(cfg.MQTT.Handler.Publish.Async.AtMostOnce),
 			mqtthandler.WithPublishAsyncAtLeastOnce(cfg.MQTT.Handler.Publish.Async.AtLeastOnce),
 			mqtthandler.WithPublishAsyncExactlyOnce(cfg.MQTT.Handler.Publish.Async.ExactlyOnce),
+			mqtthandler.WithAuthenticator(authenticator),
 		)
 
 		srv := mqttserver.New(logger, registry, httpProbe,

pkg/auth/instrument/instrument.go

@@ -0,0 +1,62 @@
+package instrument
+
+import (
+	"context"
+	"strconv"
+	"time"
+
+	"github.com/grepplabs/mqtt-proxy/apis"
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/client_golang/prometheus/promauto"
+)
+
+type authenticator struct {
+	delegate apis.UserPasswordAuthenticator
+	metrics  *authenticatorMetrics
+}
+
+type authenticatorMetrics struct {
+	loginDuration *prometheus.HistogramVec
+}
+
+func New(delegate apis.UserPasswordAuthenticator, registry *prometheus.Registry) apis.UserPasswordAuthenticator {
+	return &authenticator{
+		delegate: delegate,
+		metrics:  newAuthenticatorMetrics(delegate.Name(), registry),
+	}
+}
+
+func (p *authenticator) Name() string {
+	return p.delegate.Name()
+}
+
+func (p *authenticator) Login(ctx context.Context, authRequest *apis.UserPasswordAuthRequest) (*apis.UserPasswordAuthResponse, error) {
+	start := time.Now()
+	authResponse, err := p.delegate.Login(ctx, authRequest)
+	code := ""
+	if authResponse != nil {
+		code = strconv.Itoa(int(authResponse.ReturnCode))
+	}
+	isError := "0"
+	if err != nil {
+		isError = "1"
+	}
+	p.metrics.loginDuration.WithLabelValues(code, isError).Observe(time.Since(start).Seconds())
+	return authResponse, err
+}
+
+func (p *authenticator) Close() error {
+	return p.delegate.Close()
+}
+
+func newAuthenticatorMetrics(name string, registry *prometheus.Registry) *authenticatorMetrics {
+	loginDuration := promauto.With(registry).NewHistogramVec(prometheus.HistogramOpts{
+		Name:        "mqtt_proxy_authenticator_login_duration_seconds",
+		Help:        "Tracks the latencies for auth requests.",
+		ConstLabels: prometheus.Labels{"name": name},
+	}, []string{"code", "error"})
+
+	return &authenticatorMetrics{
+		loginDuration: loginDuration,
+	}
+}

pkg/auth/noop/noop.go

@@ -0,0 +1,37 @@
+package noop
+
+import (
+	"context"
+
+	"github.com/grepplabs/mqtt-proxy/apis"
+	"github.com/grepplabs/mqtt-proxy/pkg/log"
+	"github.com/prometheus/client_golang/prometheus"
+)
+
+const (
+	authName = "noop"
+)
+
+type noopAuthenticator struct {
+	logger log.Logger
+}
+
+func New(logger log.Logger, _ *prometheus.Registry) apis.UserPasswordAuthenticator {
+	return &noopAuthenticator{
+		logger: logger.WithField("authenticator", authName),
+	}
+}
+
+func (p noopAuthenticator) Login(_ context.Context, _ *apis.UserPasswordAuthRequest) (*apis.UserPasswordAuthResponse, error) {
+	return &apis.UserPasswordAuthResponse{
+		ReturnCode: apis.AuthAccepted,
+	}, nil
+}
+
+func (p noopAuthenticator) Close() error {
+	return nil
+}
+
+func (p noopAuthenticator) Name() string {
+	return authName
+}