Add disable_superuser options at the plugin level and as local option…

…s for relevant backends (DBs don't need it as they may have empty superuser queries). TODO: tests and docs.

backends/grpc.go

@@ -19,8 +19,9 @@ import (
 
 // GRPC holds a client for the service and implements the Backend interface.
 type GRPC struct {
- client gs.AuthServiceClient
- conn *grpc.ClientConn
+ client gs.AuthServiceClient
+ conn *grpc.ClientConn
+ disableSuperuser bool
 }
 
 // NewGRPC tries to connect to the gRPC service at the given host.
@@ -31,6 +32,10 @@ func NewGRPC(authOpts map[string]string, logLevel log.Level) (GRPC, error) {
  return g, errors.New("grpc must have a host and port")
  }
 
+ if authOpts["grpc_disable_superuser"] == "true" {
+ g.disableSuperuser = true
+ }
+
  caCert := []byte(authOpts["grpc_ca_cert"])
  tlsCert := []byte(authOpts["grpc_tls_cert"])
  tlsKey := []byte(authOpts["grpc_tls_key"])
@@ -70,6 +75,10 @@ func (o GRPC) GetUser(username, password, clientid string) bool {
 // GetSuperuser checks that the user is a superuser.
 func (o GRPC) GetSuperuser(username string) bool {
 
+ if o.disableSuperuser {
+ return false
+ }
+
  req := gs.GetSuperuserRequest{
  Username: username,
  }

backends/http.go

@@ -17,16 +17,17 @@ import (
 )
 
 type HTTP struct {
- UserUri string
- SuperuserUri string
- AclUri string
- Host string
- Port string
- WithTLS bool
- VerifyPeer bool
- ParamsMode string
- ResponseMode string
- Client *h.Client
+ UserUri string
+ SuperuserUri string
+ AclUri string
+ Host string
+ Port string
+ WithTLS bool
+ VerifyPeer bool
+ ParamsMode string
+ ResponseMode string
+ Client *h.Client
+ disableSuperuser bool
 }
 
 type HTTPResponse struct {
@@ -51,6 +52,10 @@ func NewHTTP(authOpts map[string]string, logLevel log.Level) (HTTP, error) {
  missingOpts := ""
  httpOk := true
 
+ if authOpts["http_disable_superuser"] == "true" {
+ http.disableSuperuser = true
+ }
+
  if responseMode, ok := authOpts["http_response_mode"]; ok {
  if responseMode == "text" || responseMode == "json" {
  http.ResponseMode = responseMode
@@ -144,6 +149,10 @@ func (o HTTP) GetUser(username, password, clientid string) bool {
 
 func (o HTTP) GetSuperuser(username string) bool {
 
+ if o.disableSuperuser {
+ return false
+ }
+
  var dataMap = map[string]interface{}{
  "username": username,
  }
@@ -195,15 +204,17 @@ func (o HTTP) httpRequest(uri, username string, dataMap map[string]interface{},
  if o.ParamsMode == "form" {
  resp, err = o.Client.PostForm(fullUri, urlValues)
  } else {
- dataJson, err := json.Marshal(dataMap)
+ var dataJson []byte
+ dataJson, err = json.Marshal(dataMap)
 
  if err != nil {
  log.Errorf("marshal error: %s", err)
  return false
  }
 
  contentReader := bytes.NewReader(dataJson)
- req, err := h.NewRequest("POST", fullUri, contentReader)
+ var req *h.Request
+ req, err = h.NewRequest("POST", fullUri, contentReader)
 
  if err != nil {
  log.Errorf("req error: %s", err)

backends/jwt.go

@@ -42,7 +42,8 @@ type JWT struct {
  ParamsMode string
  ResponseMode string
 
- UserField string
+ UserField string
+ disableSuperuser bool
 }
 
 // Claims defines the struct containing the token claims. StandardClaim's Subject field should contain the username, unless an opt is set to support Username field.
@@ -72,6 +73,10 @@ func NewJWT(authOpts map[string]string, logLevel log.Level) (JWT, error) {
  UserField: "Subject",
  }
 
+ if authOpts["jwt_disable_superuser"] == "true" {
+ jwt.disableSuperuser = true
+ }
+
  if userField, ok := authOpts["jwt_userfield"]; ok && userField == "Username" {
  jwt.UserField = userField
  } else {
@@ -236,7 +241,9 @@ func (o JWT) GetUser(token, password, clientid string) bool {
 
 //GetSuperuser checks if the given user is a superuser.
 func (o JWT) GetSuperuser(token string) bool {
-
+ if o.disableSuperuser {
+ return false
+ }
  if o.Remote {
  var dataMap map[string]interface{}
  var urlValues = url.Values{}

backends/mongo.go

@@ -18,14 +18,15 @@ import (
 )
 
 type Mongo struct {
- Host string
- Port string
- Username string
- Password string
- DBName string
- UsersCollection string
- AclsCollection string
- Conn *mongo.Client
+ Host string
+ Port string
+ Username string
+ Password string
+ DBName string
+ UsersCollection string
+ AclsCollection string
+ Conn *mongo.Client
+ disableSuperuser bool
 }
 
 type MongoAcl struct {
@@ -54,6 +55,10 @@ func NewMongo(authOpts map[string]string, logLevel log.Level) (Mongo, error) {
  AclsCollection: "acls",
  }
 
+ if authOpts["mongo_disable_superuser"] == "true" {
+ m.disableSuperuser = true
+ }
+
  if mongoHost, ok := authOpts["mongo_host"]; ok {
  m.Host = mongoHost
  }
@@ -135,6 +140,10 @@ func (o Mongo) GetUser(username, password, clientid string) bool {
 //GetSuperuser checks that the key username:su exists and has value "true".
 func (o Mongo) GetSuperuser(username string) bool {
 
+ if o.disableSuperuser {
+ return false
+ }
+
  uc := o.Conn.Database(o.DBName).Collection(o.UsersCollection)
 
  var user MongoUser

backends/mysql.go

@@ -182,10 +182,13 @@ func NewMysql(authOpts map[string]string, logLevel log.Level) (Mysql, error) {
  }
  clientCert = append(clientCert, certs)
 
- mq.RegisterTLSConfig("custom", &tls.Config{
+ err = mq.RegisterTLSConfig("custom", &tls.Config{
  RootCAs: rootCertPool,
  Certificates: clientCert,
  })
+ if err != nil {
+ return mysql, errors.Errorf("Mysql register TLS config error: %s", err)
+ }
  }
 
  var err error

backends/redis.go

@@ -14,11 +14,12 @@ import (
 )
 
 type Redis struct {
- Host string
- Port string
- Password string
- DB int32
- Conn *goredis.Client
+ Host string
+ Port string
+ Password string
+ DB int32
+ Conn *goredis.Client
+ disableSuperuser bool
 }
 
 func NewRedis(authOpts map[string]string, logLevel log.Level) (Redis, error) {
@@ -31,6 +32,10 @@ func NewRedis(authOpts map[string]string, logLevel log.Level) (Redis, error) {
  DB: 1,
  }
 
+ if authOpts["redis_disable_superuser"] == "true" {
+ redis.disableSuperuser = true
+ }
+
  if redisHost, ok := authOpts["redis_host"]; ok {
  redis.Host = redisHost
  }
@@ -95,6 +100,10 @@ func (o Redis) GetUser(username, password, clientid string) bool {
 //GetSuperuser checks that the key username:su exists and has value "true".
 func (o Redis) GetSuperuser(username string) bool {
 
+ if o.disableSuperuser {
+ return false
+ }
+
  isSuper, err := o.Conn.Get(fmt.Sprintf("%s:su", username)).Result()
 
  if err != nil {

go-auth.go

@@ -45,6 +45,7 @@ type CommonData struct {
  LogLevel log.Level
  LogDest string
  LogFile string
+ disableSuperuser bool
 }
 
 //Cache stores necessary values for Redis cache
@@ -125,6 +126,11 @@ func AuthPluginInit(keys []string, values []string, authOptsNum int) {
  log.Fatal("backends error")
  }
 
+ //Disable superusers for all backends if option is set.
+ if authOpts["disable_superuser"] == "true" {
+ commonData.disableSuperuser = true
+ }
+
  //Check if log level is given. Set level if any valid option is given.
  if logLevel, ok := authOpts["log_level"]; ok {
 
@@ -518,7 +524,7 @@ func AuthAclCheck(clientid, username, topic string, acc int) bool {
  }
  }
 
- //If prefixes are enabled, checkt if username has a valid prefix and use the correct backend if so.
+ //If prefixes are enabled, check if username has a valid prefix and use the correct backend if so.
  //Else, check all backends.
  if commonData.CheckPrefix {
  validPrefix, bename := CheckPrefix(username)
@@ -533,7 +539,8 @@ func AuthAclCheck(clientid, username, topic string, acc int) bool {
  var backend = commonData.Backends[bename]
 
  log.Debugf("Superuser check with backend %s", backend.GetName())
- if backend.GetSuperuser(username) {
+ // Short circuit checks when superusers are disabled.
+ if !commonData.disableSuperuser && backend.GetSuperuser(username) {
  log.Debugf("superuser %s acl authenticated with backend %s", username, backend.GetName())
  aclCheck = true
  }
@@ -691,7 +698,8 @@ func CheckBackendsAcl(username, topic, clientid string, acc int) bool {
  var backend = commonData.Backends[bename]
 
  log.Debugf("Superuser check with backend %s", backend.GetName())
- if backend.GetSuperuser(username) {
+ // Short circuit checks when superusers are disabled.
+ if !commonData.disableSuperuser && backend.GetSuperuser(username) {
  log.Debugf("superuser %s acl authenticated with backend %s", username, backend.GetName())
  aclCheck = true
  break