add subscription token based auth

auth/auth.go

@@ -10,6 +10,9 @@ import (
 type Interface interface {
 	Authenticate(username, password string) (User, error)
 	User(username string) (User, error)
+	SubscriptionUser(token string) (User, error)
+	GrantSubscriptionAccess(token, db, rp string) error
+	RevokeSubscriptionAccess(token string) error
 }
 
 // ErrAuthenticate is returned when authentication fails.

server/server.go

@@ -142,13 +142,13 @@ func New(c *Config, buildInfo BuildInfo, logService logging.Interface) (*Server,
 	s.appendDeadmanService()
 	s.appendSMTPService()
 	s.InitHTTPDService()
+	s.appendStorageService()
+	s.appendAuthService()
 	if err := s.appendInfluxDBService(); err != nil {
 		return nil, errors.Wrap(err, "influxdb service")
 	}
-	s.appendStorageService()
 	s.appendTaskStoreService()
 	s.appendReplayService()
-	s.appendAuthService()
 
 	// Append Alert integration services
 	s.appendOpsGenieService()
@@ -215,9 +215,10 @@ func (s *Server) appendInfluxDBService() error {
 		if err != nil {
 			return errors.Wrap(err, "failed to get http port")
 		}
-		srv := influxdb.NewService(c, s.config.defaultInfluxDB, httpPort, s.config.Hostname, l)
+		srv := influxdb.NewService(c, s.config.defaultInfluxDB, httpPort, s.config.Hostname, s.config.HTTP.AuthEnabled, l)
 		srv.PointsWriter = s.TaskMaster
 		srv.LogService = s.LogService
+		srv.AuthService = s.AuthService
 
 		s.InfluxDBService = srv
 		s.TaskMaster.InfluxDBService = srv

services/httpd/handler.go

@@ -45,6 +45,7 @@ type AuthenticationMethod int
 const (
 	UserAuthentication AuthenticationMethod = iota
 	BearerAuthentication
+	SubscriptionAuthentication
 )
 
 type AuthorizationHandler func(http.ResponseWriter, *http.Request, auth.User)
@@ -612,6 +613,11 @@ func authenticate(inner AuthorizationHandler, h *Handler, requireAuthentication
 				HttpError(w, err.Error(), false, http.StatusUnauthorized)
 				return
 			}
+		case SubscriptionAuthentication:
+			if user, err = h.AuthService.SubscriptionUser(creds.Token); err != nil {
+				HttpError(w, err.Error(), false, http.StatusUnauthorized)
+				return
+			}
 		default:
 			HttpError(w, "unsupported authentication", false, http.StatusUnauthorized)
 		}
@@ -684,6 +690,7 @@ type credentials struct {
 // As params: http://127.0.0.1/query?u=username&p=password
 // As basic auth: http://username:password@127.0.0.1
 // As Bearer token in Authorization header: Bearer <JWT_TOKEN_BLOB>
+// As simple acccess token in InfluxDB-Access-Token: <TOKEN>
 func parseCredentials(r *http.Request) (credentials, error) {
 	q := r.URL.Query()
 
@@ -706,6 +713,13 @@ func parseCredentials(r *http.Request) (credentials, error) {
 				Password: p,
 			}, nil
 		}
+	} else if s := r.Header.Get("InfluxDB-Access-Token"); s != "" {
+		// Check for the HTTP InfluxDB-Access-Token header.
+		return credentials{
+			Method: SubscriptionAuthentication,
+			Token:  s,
+		}, nil
+
 	}
 
 	// Check for username and password in URL params.

services/influxdb/service.go

@@ -2,10 +2,12 @@ package influxdb
 
 import (
 	"bytes"
+	"crypto/rand"
 	"crypto/tls"
 	"crypto/x509"
-	"errors"
+	"encoding/base64"
 	"fmt"
+	"io"
 	"io/ioutil"
 	"log"
 	"net"
@@ -21,12 +23,16 @@ import (
 	"github.com/influxdata/influxdb/models"
 	"github.com/influxdata/kapacitor"
 	"github.com/influxdata/kapacitor/services/udp"
+	"github.com/pkg/errors"
 )
 
 const (
 	// Legacy name given to all subscriptions.
 	legacySubName = "kapacitor"
 	subNamePrefix = "kapacitor-"
+
+	// Size in bytes of a token for subscription authentication
+	tokenSize = 64
 )
 
 // Handles requests to write or read from an InfluxDB cluster
@@ -40,10 +46,15 @@ type Service struct {
 	LogService interface {
 		NewLogger(string, int) *log.Logger
 	}
+
+	AuthService interface {
+		GrantSubscriptionAccess(token, db, rp string) error
+		RevokeSubscriptionAccess(token string) error
+	}
 	logger *log.Logger
 }
 
-func NewService(configs []Config, defaultInfluxDB, httpPort int, hostname string, l *log.Logger) *Service {
+func NewService(configs []Config, defaultInfluxDB, httpPort int, hostname string, useTokens bool, l *log.Logger) *Service {
 	clusterID := kapacitor.ClusterIDVar.StringValue()
 	subName := subNamePrefix + clusterID
 	clusters := make(map[string]*influxdb, len(configs))
@@ -100,6 +111,7 @@ func NewService(configs []Config, defaultInfluxDB, httpPort int, hostname string
 			disableSubs:              c.DisableSubscriptions,
 			protocol:                 c.SubscriptionProtocol,
 			runningSubs:              runningSubs,
+			useTokens:                useTokens,
 		}
 		if defaultInfluxDB == i {
 			defaultInfluxDBName = c.Name
@@ -116,6 +128,7 @@ func (s *Service) Open() error {
 	for _, cluster := range s.clusters {
 		cluster.PointsWriter = s.PointsWriter
 		cluster.LogService = s.LogService
+		cluster.AuthService = s.AuthService
 		err := cluster.Open()
 		if err != nil {
 			return err
@@ -164,6 +177,7 @@ type influxdb struct {
 	subscriptionSyncInterval time.Duration
 	disableSubs              bool
 	runningSubs              map[subEntry]bool
+	useTokens                bool
 
 	clusterID     string
 	subName       string
@@ -175,6 +189,10 @@ type influxdb struct {
 	LogService interface {
 		NewLogger(string, int) *log.Logger
 	}
+	AuthService interface {
+		GrantSubscriptionAccess(token, db, rp string) error
+		RevokeSubscriptionAccess(token string) error
+	}
 
 	services []interface {
 		Open() error
@@ -388,7 +406,12 @@ func (s *influxdb) linkSubscriptions() error {
 					// Check if the hostname, port or protocol have changed
 					if host != s.hostname ||
 						u.Scheme != s.protocol ||
-						((u.Scheme == "http" || u.Scheme == "https") && int(pn) != s.httpPort) {
+						((u.Scheme == "http" || u.Scheme == "https") && int(pn) != s.httpPort) ||
+						(s.useTokens && u.User == nil) {
+						// Remove access for changing subscriptions.
+						if u.User != nil {
+							s.AuthService.RevokeSubscriptionAccess(u.User.Username())
+						}
 						// Something changed, drop the sub and let it get recreated
 						s.dropSub(cli, se.name, se.cluster, se.rp)
 					} else {
@@ -434,7 +457,20 @@ func (s *influxdb) linkSubscriptions() error {
 			var destination string
 			switch s.protocol {
 			case "http", "https":
-				destination = fmt.Sprintf("%s://%s:%d", s.protocol, s.hostname, s.httpPort)
+				if s.useTokens {
+					// Generate token
+					token, err := s.generateRandomToken()
+					if err != nil {
+						return errors.Wrap(err, "generating token")
+					}
+					err = s.AuthService.GrantSubscriptionAccess(token, se.cluster, se.rp)
+					if err != nil {
+						return err
+					}
+					destination = fmt.Sprintf("%s://%s@%s:%d", s.protocol, token, s.hostname, s.httpPort)
+				} else {
+					destination = fmt.Sprintf("%s://%s:%d", s.protocol, s.hostname, s.httpPort)
+				}
 			case "udp":
 				addr, err := s.startUDPListener(se.cluster, se.rp, "0")
 				if err != nil {
@@ -457,6 +493,14 @@ func (s *influxdb) linkSubscriptions() error {
 	return nil
 }
 
+func (s *influxdb) generateRandomToken() (string, error) {
+	tokenBytes := make([]byte, tokenSize)
+	if _, err := io.ReadFull(rand.Reader, tokenBytes); err != nil {
+		return "", err
+	}
+	return base64.RawURLEncoding.EncodeToString(tokenBytes), nil
+}
+
 func (s *influxdb) createSub(cli client.Client, name, cluster, rp, mode string, destinations []string) (err error) {
 	var buf bytes.Buffer
 	for i, dst := range destinations {

services/noauth/service.go

@@ -37,3 +37,17 @@ func (s *Service) User(username string) (auth.User, error) {
 	s.logger.Println("W! using noauth auth backend. Faked authentication for user", username)
 	return auth.NewUser(username, nil, true, nil), nil
 }
+
+// Return a user will all privileges.
+func (s *Service) SubscriptionUser(token string) (auth.User, error) {
+	s.logger.Println("W! using noauth auth backend. Faked authentication for subscription user token")
+	return auth.NewUser("subscription-user", nil, true, nil), nil
+}
+
+func (s *Service) GrantSubscriptionAccess(token, db, rp string) error {
+	return nil
+}
+
+func (s *Service) RevokeSubscriptionAccess(token string) error {
+	return nil
+}