MF-1257 - Access messages from readers endpoint with user access token

api/openapi/readers.yml

@@ -109,7 +109,10 @@ components:
   parameters:
     Authorization:
       name: Authorization
-      description: Thing access token.
+      description: | 
+        Thing or User access token: 
+        * For thing access use "Authorization: Thing <thing_key>"
+        * For user access use "Authorization: Bearer <user_token>"
       in: header
       schema:
         type: string

cmd/cassandra-reader/main.go

@@ -19,6 +19,7 @@ import (
 	kitprometheus "github.com/go-kit/kit/metrics/prometheus"
 	"github.com/gocql/gocql"
 	"github.com/mainflux/mainflux"
+	authapi "github.com/mainflux/mainflux/auth/api/grpc"
 	"github.com/mainflux/mainflux/logger"
 	"github.com/mainflux/mainflux/readers"
 	"github.com/mainflux/mainflux/readers/api"
@@ -46,8 +47,10 @@ const (
 	defServerCert        = ""
 	defServerKey         = ""
 	defJaegerURL         = ""
-	defThingsAuthURL     = "localhost:8181"
+	defThingsAuthURL     = "localhost:8183"
 	defThingsAuthTimeout = "1s"
+	defUsersAuthURL      = "localhost:8181"
+	defUsersAuthTimeout  = "1s"
 
 	envLogLevel          = "MF_CASSANDRA_READER_LOG_LEVEL"
 	envPort              = "MF_CASSANDRA_READER_PORT"
@@ -63,6 +66,8 @@ const (
 	envJaegerURL         = "MF_JAEGER_URL"
 	envThingsAuthURL     = "MF_THINGS_AUTH_GRPC_URL"
 	envThingsAuthTimeout = "MF_THINGS_AUTH_GRPC_TIMEOUT"
+	envUsersAuthURL      = "MF_AUTH_GRPC_URL"
+	envUsersAuthTimeout  = "MF_AUTH_GRPC_TIMEOUT"
 )
 
 type config struct {
@@ -75,7 +80,9 @@ type config struct {
 	serverKey         string
 	jaegerURL         string
 	thingsAuthURL     string
+	usersAuthURL      string
 	thingsAuthTimeout time.Duration
+	usersAuthTimeout  time.Duration
 }
 
 func main() {
@@ -96,11 +103,19 @@ func main() {
 	defer thingsCloser.Close()
 
 	tc := thingsapi.NewClient(conn, thingsTracer, cfg.thingsAuthTimeout)
+	authTracer, authCloser := initJaeger("auth", cfg.jaegerURL, logger)
+	defer authCloser.Close()
+
+	authConn := connectToAuth(cfg, logger)
+	defer authConn.Close()
+
+	auth := authapi.NewClient(authTracer, authConn, cfg.usersAuthTimeout)
+
 	repo := newService(session, logger)
 
 	errs := make(chan error, 2)
 
-	go startHTTPServer(repo, tc, cfg, errs, logger)
+	go startHTTPServer(repo, tc, auth, cfg, errs, logger)
 
 	go func() {
 		c := make(chan os.Signal)
@@ -112,6 +127,32 @@ func main() {
 	logger.Error(fmt.Sprintf("Cassandra reader service terminated: %s", err))
 }
 
+func connectToAuth(cfg config, logger logger.Logger) *grpc.ClientConn {
+	var opts []grpc.DialOption
+	logger.Info("Connecting to auth via gRPC")
+	if cfg.clientTLS {
+		if cfg.caCerts != "" {
+			tpc, err := credentials.NewClientTLSFromFile(cfg.caCerts, "")
+			if err != nil {
+				logger.Error(fmt.Sprintf("Failed to create tls credentials: %s", err))
+				os.Exit(1)
+			}
+			opts = append(opts, grpc.WithTransportCredentials(tpc))
+		}
+	} else {
+		opts = append(opts, grpc.WithInsecure())
+		logger.Info("gRPC communication is not encrypted")
+	}
+
+	conn, err := grpc.Dial(cfg.usersAuthURL, opts...)
+	if err != nil {
+		logger.Error(fmt.Sprintf("Failed to connect to auth service: %s", err))
+		os.Exit(1)
+	}
+	logger.Info(fmt.Sprintf("Established gRPC connection to things via gRPC: %s", cfg.usersAuthURL))
+	return conn
+}
+
 func loadConfig() config {
 	dbPort, err := strconv.Atoi(mainflux.Env(envDBPort, defDBPort))
 	if err != nil {
@@ -136,6 +177,11 @@ func loadConfig() config {
 		log.Fatalf("Invalid %s value: %s", envThingsAuthTimeout, err.Error())
 	}
 
+	usersAuthTimeout, err := time.ParseDuration(mainflux.Env(envUsersAuthTimeout, defUsersAuthTimeout))
+	if err != nil {
+		log.Fatalf("Invalid %s value: %s", envThingsAuthTimeout, err.Error())
+	}
+
 	return config{
 		logLevel:          mainflux.Env(envLogLevel, defLogLevel),
 		port:              mainflux.Env(envPort, defPort),
@@ -146,6 +192,8 @@ func loadConfig() config {
 		serverKey:         mainflux.Env(envServerKey, defServerKey),
 		jaegerURL:         mainflux.Env(envJaegerURL, defJaegerURL),
 		thingsAuthURL:     mainflux.Env(envThingsAuthURL, defThingsAuthURL),
+		usersAuthURL:      mainflux.Env(envUsersAuthURL, defUsersAuthURL),
+		usersAuthTimeout:  usersAuthTimeout,
 		thingsAuthTimeout: authTimeout,
 	}
 }
@@ -181,6 +229,7 @@ func connectToThings(cfg config, logger logger.Logger) *grpc.ClientConn {
 		logger.Error(fmt.Sprintf("Failed to connect to things service: %s", err))
 		os.Exit(1)
 	}
+	logger.Info(fmt.Sprintf("Established gRPC connection to things via gRPC: %s", cfg.thingsAuthURL))
 	return conn
 }
 
@@ -230,14 +279,14 @@ func newService(session *gocql.Session, logger logger.Logger) readers.MessageRep
 	return repo
 }
 
-func startHTTPServer(repo readers.MessageRepository, tc mainflux.ThingsServiceClient, cfg config, errs chan error, logger logger.Logger) {
+func startHTTPServer(repo readers.MessageRepository, tc mainflux.ThingsServiceClient, ac mainflux.AuthServiceClient, cfg config, errs chan error, logger logger.Logger) {
 	p := fmt.Sprintf(":%s", cfg.port)
 	if cfg.serverCert != "" || cfg.serverKey != "" {
 		logger.Info(fmt.Sprintf("Cassandra reader service started using https on port %s with cert %s key %s",
 			cfg.port, cfg.serverCert, cfg.serverKey))
-		errs <- http.ListenAndServeTLS(p, cfg.serverCert, cfg.serverKey, api.MakeHandler(repo, tc, "cassandra-reader"))
+		errs <- http.ListenAndServeTLS(p, cfg.serverCert, cfg.serverKey, api.MakeHandler(repo, tc, ac, "cassandra-reader"))
 		return
 	}
 	logger.Info(fmt.Sprintf("Cassandra reader service started, exposed port %s", cfg.port))
-	errs <- http.ListenAndServe(p, api.MakeHandler(repo, tc, "cassandra-reader"))
+	errs <- http.ListenAndServe(p, api.MakeHandler(repo, tc, ac, "cassandra-reader"))
 }

cmd/coap/main.go

@@ -37,7 +37,7 @@ const (
 	defClientTLS         = "false"
 	defCACerts           = ""
 	defJaegerURL         = ""
-	defThingsAuthURL     = "localhost:8181"
+	defThingsAuthURL     = "localhost:8183"
 	defThingsAuthTimeout = "1s"
 
 	envPort              = "MF_COAP_ADAPTER_PORT"

cmd/http/main.go

@@ -37,7 +37,7 @@ const (
 	defPort              = "8180"
 	defNatsURL           = "nats://localhost:4222"
 	defJaegerURL         = ""
-	defThingsAuthURL     = "localhost:8181"
+	defThingsAuthURL     = "localhost:8183"
 	defThingsAuthTimeout = "1s"
 
 	envLogLevel          = "MF_HTTP_ADAPTER_LOG_LEVEL"

cmd/influxdb-reader/main.go

@@ -15,6 +15,7 @@ import (
 	kitprometheus "github.com/go-kit/kit/metrics/prometheus"
 	influxdata "github.com/influxdata/influxdb/client/v2"
 	"github.com/mainflux/mainflux"
+	authapi "github.com/mainflux/mainflux/auth/api/grpc"
 	"github.com/mainflux/mainflux/logger"
 	"github.com/mainflux/mainflux/readers"
 	"github.com/mainflux/mainflux/readers/api"
@@ -40,8 +41,10 @@ const (
 	defServerCert        = ""
 	defServerKey         = ""
 	defJaegerURL         = ""
-	defThingsAuthURL     = "localhost:8181"
+	defThingsAuthURL     = "localhost:8183"
 	defThingsAuthTimeout = "1s"
+	defUsersAuthURL      = "localhost:8181"
+	defUsersAuthTimeout  = "1s"
 
 	envLogLevel          = "MF_INFLUX_READER_LOG_LEVEL"
 	envPort              = "MF_INFLUX_READER_PORT"
@@ -57,6 +60,8 @@ const (
 	envJaegerURL         = "MF_JAEGER_URL"
 	envThingsAuthURL     = "MF_THINGS_AUTH_GRPC_URL"
 	envThingsAuthTimeout = "MF_THINGS_AUTH_GRPC_TIMEOUT"
+	envAuthURL           = "MF_AUTH_GRPC_URL"
+	envUsersAuthTimeout  = "MF_AUTH_GRPC_TIMEOUT"
 )
 
 type config struct {
@@ -73,7 +78,9 @@ type config struct {
 	serverKey         string
 	jaegerURL         string
 	thingsAuthURL     string
+	usersAuthURL      string
 	thingsAuthTimeout time.Duration
+	usersAuthTimeout  time.Duration
 }
 
 func main() {
@@ -90,6 +97,14 @@ func main() {
 
 	tc := thingsapi.NewClient(conn, thingsTracer, cfg.thingsAuthTimeout)
 
+	authTracer, authCloser := initJaeger("auth", cfg.jaegerURL, logger)
+	defer authCloser.Close()
+
+	authConn := connectToAuth(cfg, logger)
+	defer authConn.Close()
+
+	auth := authapi.NewClient(authTracer, authConn, cfg.usersAuthTimeout)
+
 	client, err := influxdata.NewHTTPClient(clientCfg)
 	if err != nil {
 		logger.Error(fmt.Sprintf("Failed to create InfluxDB client: %s", err))
@@ -106,12 +121,38 @@ func main() {
 		errs <- fmt.Errorf("%s", <-c)
 	}()
 
-	go startHTTPServer(repo, tc, cfg, logger, errs)
+	go startHTTPServer(repo, tc, auth, cfg, logger, errs)
 
 	err = <-errs
 	logger.Error(fmt.Sprintf("InfluxDB writer service terminated: %s", err))
 }
 
+func connectToAuth(cfg config, logger logger.Logger) *grpc.ClientConn {
+	var opts []grpc.DialOption
+	logger.Info("Connecting to auth via gRPC")
+	if cfg.clientTLS {
+		if cfg.caCerts != "" {
+			tpc, err := credentials.NewClientTLSFromFile(cfg.caCerts, "")
+			if err != nil {
+				logger.Error(fmt.Sprintf("Failed to create tls credentials: %s", err))
+				os.Exit(1)
+			}
+			opts = append(opts, grpc.WithTransportCredentials(tpc))
+		}
+	} else {
+		opts = append(opts, grpc.WithInsecure())
+		logger.Info("gRPC communication is not encrypted")
+	}
+
+	conn, err := grpc.Dial(cfg.usersAuthURL, opts...)
+	if err != nil {
+		logger.Error(fmt.Sprintf("Failed to connect to auth service: %s", err))
+		os.Exit(1)
+	}
+	logger.Info("Established gRPC connection to auth via gRPC")
+	return conn
+}
+
 func loadConfigs() (config, influxdata.HTTPConfig) {
 	tls, err := strconv.ParseBool(mainflux.Env(envClientTLS, defClientTLS))
 	if err != nil {
@@ -123,6 +164,11 @@ func loadConfigs() (config, influxdata.HTTPConfig) {
 		log.Fatalf("Invalid %s value: %s", envThingsAuthTimeout, err.Error())
 	}
 
+	userAuthTimeout, err := time.ParseDuration(mainflux.Env(envUsersAuthTimeout, defUsersAuthTimeout))
+	if err != nil {
+		log.Fatalf("Invalid %s value: %s", envThingsAuthTimeout, err.Error())
+	}
+
 	cfg := config{
 		logLevel:          mainflux.Env(envLogLevel, defLogLevel),
 		port:              mainflux.Env(envPort, defPort),
@@ -138,6 +184,8 @@ func loadConfigs() (config, influxdata.HTTPConfig) {
 		jaegerURL:         mainflux.Env(envJaegerURL, defJaegerURL),
 		thingsAuthURL:     mainflux.Env(envThingsAuthURL, defThingsAuthURL),
 		thingsAuthTimeout: authTimeout,
+		usersAuthURL:      mainflux.Env(envAuthURL, defUsersAuthURL),
+		usersAuthTimeout:  userAuthTimeout,
 	}
 
 	clientCfg := influxdata.HTTPConfig{
@@ -151,6 +199,7 @@ func loadConfigs() (config, influxdata.HTTPConfig) {
 
 func connectToThings(cfg config, logger logger.Logger) *grpc.ClientConn {
 	var opts []grpc.DialOption
+	logger.Info("connecting to things via gRPC")
 	if cfg.clientTLS {
 		if cfg.caCerts != "" {
 			tpc, err := credentials.NewClientTLSFromFile(cfg.caCerts, "")
@@ -170,6 +219,7 @@ func connectToThings(cfg config, logger logger.Logger) *grpc.ClientConn {
 		logger.Error(fmt.Sprintf("Failed to connect to things service: %s", err))
 		os.Exit(1)
 	}
+	logger.Info(fmt.Sprintf("Established gRPC connection to things via gRPC: %s", cfg.thingsAuthURL))
 	return conn
 }
 
@@ -219,14 +269,14 @@ func newService(client influxdata.Client, dbName string, logger logger.Logger) r
 	return repo
 }
 
-func startHTTPServer(repo readers.MessageRepository, tc mainflux.ThingsServiceClient, cfg config, logger logger.Logger, errs chan error) {
+func startHTTPServer(repo readers.MessageRepository, tc mainflux.ThingsServiceClient, ac mainflux.AuthServiceClient, cfg config, logger logger.Logger, errs chan error) {
 	p := fmt.Sprintf(":%s", cfg.port)
 	if cfg.serverCert != "" || cfg.serverKey != "" {
 		logger.Info(fmt.Sprintf("InfluxDB reader service started using https on port %s with cert %s key %s",
 			cfg.port, cfg.serverCert, cfg.serverKey))
-		errs <- http.ListenAndServeTLS(p, cfg.serverCert, cfg.serverKey, api.MakeHandler(repo, tc, "influxdb-reader"))
+		errs <- http.ListenAndServeTLS(p, cfg.serverCert, cfg.serverKey, api.MakeHandler(repo, tc, ac, "influxdb-reader"))
 		return
 	}
 	logger.Info(fmt.Sprintf("InfluxDB reader service started, exposed port %s", cfg.port))
-	errs <- http.ListenAndServe(p, api.MakeHandler(repo, tc, "influxdb-reader"))
+	errs <- http.ListenAndServe(p, api.MakeHandler(repo, tc, ac, "influxdb-reader"))
 }