Fix #471. User should be able to update permissions

CHANGELOG.md

@@ -1,3 +1,9 @@
+## v0.5.11 [unreleased]
+
+### Features
+
+- [Issue #471](https://github.com/influxdb/influxdb/issues/471). Read and write permissions should be settable through the http api
+
 ## v0.5.10 [2014-04-22]
 
 ### Features

src/api/http/api.go

@@ -549,6 +549,8 @@ type NewUser struct {
 	Name     string `json:"name"`
 	Password string `json:"password"`
 	IsAdmin  bool   `json:"isAdmin"`
+	ReadFrom string `json:"readFrom"`
+	WriteTo  string `json:"writeTo"`
 }
 
 type UpdateClusterAdminUser struct {
@@ -766,8 +768,16 @@ func (self *HttpServer) createDbUser(w libhttp.ResponseWriter, r *libhttp.Reques
 	db := r.URL.Query().Get(":db")
 
 	self.tryAsDbUserAndClusterAdmin(w, r, func(u User) (int, interface{}) {
+		permissions := []string{}
+		if newUser.ReadFrom != "" || newUser.WriteTo != "" {
+			if newUser.ReadFrom == "" || newUser.WriteTo == "" {
+				return libhttp.StatusBadRequest, "You have to provide read and write permissions"
+			}
+			permissions = append(permissions, newUser.ReadFrom, newUser.WriteTo)
+		}
+
 		username := newUser.Name
-		if err := self.userManager.CreateDbUser(u, db, username, newUser.Password); err != nil {
+		if err := self.userManager.CreateDbUser(u, db, username, newUser.Password, permissions...); err != nil {
 			log.Error("Cannot create user: %s", err)
 			return errorToStatusCode(err), err.Error()
 		}
@@ -826,6 +836,17 @@ func (self *HttpServer) updateDbUser(w libhttp.ResponseWriter, r *libhttp.Reques
 			}
 		}
 
+		if readPermissions, ok := updateUser["readFrom"]; ok {
+			writePermissions, ok := updateUser["writeTo"]
+			if !ok {
+				return libhttp.StatusBadRequest, "Changing permissions requires passing readFrom and writeTo"
+			}
+
+			if err := self.userManager.ChangeDbUserPermissions(u, db, newUser, readPermissions.(string), writePermissions.(string)); err != nil {
+				return errorToStatusCode(err), err.Error()
+			}
+		}
+
 		if admin, ok := updateUser["admin"]; ok {
 			isAdmin, ok := admin.(bool)
 			if !ok {

src/api/http/mock_user_manager_test.go

@@ -46,6 +46,7 @@ func (self MockDbUser) HasReadAccess(_ string) bool {
 }
 
 type MockUserManager struct {
+	UserManager
 	dbUsers       map[string]map[string]MockDbUser
 	clusterAdmins []string
 	ops           []*Operation
@@ -94,7 +95,7 @@ func (self *MockUserManager) ChangeClusterAdminPassword(requester common.User, u
 	return nil
 }
 
-func (self *MockUserManager) CreateDbUser(request common.User, db, username, password string) error {
+func (self *MockUserManager) CreateDbUser(request common.User, db, username, password string, permissions ...string) error {
 	if username == "" {
 		return fmt.Errorf("Invalid empty username")
 	}

src/api/http/user_manager.go

@@ -19,11 +19,12 @@ type UserManager interface {
 	// list cluster admins. only a cluster admin can list the other cluster admins
 	ListClusterAdmins(requester common.User) ([]string, error)
 	// Create a db user, it's an error if requester isn't a db admin or cluster admin
-	CreateDbUser(request common.User, db, username, password string) error
+	CreateDbUser(request common.User, db, username, password string, permissions ...string) error
 	// Delete a db user. Same restrictions apply as in CreateDbUser
 	DeleteDbUser(requester common.User, db, username string) error
 	// Change db user's password. It's an error if requester isn't a cluster admin or db admin
 	ChangeDbUserPassword(requester common.User, db, username, password string) error
+	ChangeDbUserPermissions(requester common.User, db, username, readPermissions, writePermissions string) error
 	// list cluster admins. only a cluster admin or the db admin can list the db users
 	ListDbUsers(requester common.User, db string) ([]common.User, error)
 	GetDbUser(requester common.User, db, username string) (common.User, error)

src/cluster/cluster_configuration.go

@@ -401,6 +401,20 @@ func (self *ClusterConfiguration) ChangeDbUserPassword(db, username, hash string
 	return nil
 }
 
+func (self *ClusterConfiguration) ChangeDbUserPermissions(db, username, readPermissions, writePermissions string) error {
+	self.usersLock.Lock()
+	defer self.usersLock.Unlock()
+	dbUsers := self.dbUsers[db]
+	if dbUsers == nil {
+		return fmt.Errorf("Invalid database name %s", db)
+	}
+	if dbUsers[username] == nil {
+		return fmt.Errorf("Invalid username %s", username)
+	}
+	dbUsers[username].ChangePermissions(readPermissions, writePermissions)
+	return nil
+}
+
 func (self *ClusterConfiguration) GetClusterAdmins() (names []string) {
 	self.usersLock.RLock()
 	defer self.usersLock.RUnlock()

src/cluster/user.go

@@ -1,10 +1,11 @@
 package cluster
 
 import (
-	"code.google.com/p/go.crypto/bcrypt"
 	"common"
-	"github.com/influxdb/go-cache"
 	"regexp"
+
+	"code.google.com/p/go.crypto/bcrypt"
+	"github.com/influxdb/go-cache"
 )
 
 var userCache *cache.Cache
@@ -131,6 +132,11 @@ func (self *DbUser) GetDb() string {
 	return self.Db
 }
 
+func (self *DbUser) ChangePermissions(readPermissions, writePermissions string) {
+	self.ReadFrom = []*Matcher{&Matcher{true, readPermissions}}
+	self.WriteTo = []*Matcher{&Matcher{true, writePermissions}}
+}
+
 func HashPassword(password string) ([]byte, error) {
 	if length := len(password); length < 4 || length > 56 {
 		return nil, common.NewQueryError(common.InvalidArgument, "Password must be more than 4 and less than 56 characters")

src/coordinator/command.go

@@ -21,6 +21,7 @@ func init() {
 		&SaveDbUserCommand{},
 		&SaveClusterAdminCommand{},
 		&ChangeDbUserPassword{},
+		&ChangeDbUserPermissions{},
 		&CreateContinuousQueryCommand{},
 		&DeleteContinuousQueryCommand{},
 		&SetContinuousQueryTimestampCommand{},
@@ -169,6 +170,32 @@ func (c *ChangeDbUserPassword) Apply(server raft.Server) (interface{}, error) {
 	return nil, config.ChangeDbUserPassword(c.Database, c.Username, c.Hash)
 }
 
+type ChangeDbUserPermissions struct {
+	Database         string
+	Username         string
+	ReadPermissions  string
+	WritePermissions string
+}
+
+func NewChangeDbUserPermissionsCommand(db, username, readPermissions, writePermissions string) *ChangeDbUserPermissions {
+	return &ChangeDbUserPermissions{
+		Database:         db,
+		Username:         username,
+		ReadPermissions:  readPermissions,
+		WritePermissions: writePermissions,
+	}
+}
+
+func (c *ChangeDbUserPermissions) CommandName() string {
+	return "change_db_user_password"
+}
+
+func (c *ChangeDbUserPermissions) Apply(server raft.Server) (interface{}, error) {
+	log.Debug("(raft:%s) changing db user password for %s:%s", server.Name(), c.Database, c.Username)
+	config := server.Context().(*cluster.ClusterConfiguration)
+	return nil, config.ChangeDbUserPermissions(c.Database, c.Username, c.ReadPermissions, c.WritePermissions)
+}
+
 type SaveClusterAdminCommand struct {
 	User *cluster.ClusterAdmin `json:"user"`
 }

src/coordinator/coordinator.go

@@ -126,13 +126,29 @@ func (self *CoordinatorImpl) RunQuery(user common.User, database string, querySt
 		if selectQuery.IsContinuousQuery() {
 			return self.CreateContinuousQuery(user, database, queryString)
 		}
-
+		if err := self.checkPermission(user, querySpec); err != nil {
+			return err
+		}
 		return self.runQuery(querySpec, seriesWriter)
 	}
 	seriesWriter.Close()
 	return nil
 }
 
+func (self *CoordinatorImpl) checkPermission(user common.User, querySpec *parser.QuerySpec) error {
+	// if this isn't a regex query do the permission check here
+	fromClause := querySpec.SelectQuery().GetFromClause()
+
+	for _, n := range fromClause.Names {
+		if _, ok := n.Name.GetCompiledRegex(); ok {
+			break
+		} else if name := n.Name.Name; !user.HasReadAccess(name) {
+			return fmt.Errorf("User doesn't have read access to %s", name)
+		}
+	}
+	return nil
+}
+
 // This should only get run for SelectQuery types
 func (self *CoordinatorImpl) runQuery(querySpec *parser.QuerySpec, seriesWriter SeriesWriter) error {
 	return self.runQuerySpec(querySpec, seriesWriter)
@@ -796,7 +812,7 @@ func (self *CoordinatorImpl) ChangeClusterAdminPassword(requester common.User, u
 	return self.raftServer.SaveClusterAdminUser(user)
 }
 
-func (self *CoordinatorImpl) CreateDbUser(requester common.User, db, username, password string) error {
+func (self *CoordinatorImpl) CreateDbUser(requester common.User, db, username, password string, permissions ...string) error {
 	if !requester.IsClusterAdmin() && !requester.IsDbAdmin(db) {
 		return common.NewAuthorizationError("Insufficient permissions")
 	}
@@ -818,13 +834,20 @@ func (self *CoordinatorImpl) CreateDbUser(requester common.User, db, username, p
 	if self.clusterConfiguration.GetDbUser(db, username) != nil {
 		return fmt.Errorf("User %s already exists", username)
 	}
-	matchers := []*cluster.Matcher{&cluster.Matcher{true, ".*"}}
+	readMatcher := []*cluster.Matcher{&cluster.Matcher{true, ".*"}}
+	writeMatcher := []*cluster.Matcher{&cluster.Matcher{true, ".*"}}
+	switch len(permissions) {
+	case 0:
+	case 2:
+		readMatcher[0].Name = permissions[0]
+		writeMatcher[0].Name = permissions[0]
+	}
 	log.Debug("(raft:%s) Creating user %s:%s", self.raftServer.(*RaftServer).raftServer.Name(), db, username)
 	return self.raftServer.SaveDbUser(&cluster.DbUser{cluster.CommonUser{
 		Name:     username,
 		Hash:     string(hash),
 		CacheKey: db + "%" + username,
-	}, db, matchers, matchers, false})
+	}, db, readMatcher, writeMatcher, false})
 }
 
 func (self *CoordinatorImpl) DeleteDbUser(requester common.User, db, username string) error {
@@ -873,6 +896,14 @@ func (self *CoordinatorImpl) ChangeDbUserPassword(requester common.User, db, use
 	return self.raftServer.ChangeDbUserPassword(db, username, hash)
 }
 
+func (self *CoordinatorImpl) ChangeDbUserPermissions(requester common.User, db, username, readPermissions, writePermissions string) error {
+	if !requester.IsClusterAdmin() && !requester.IsDbAdmin(db) {
+		return common.NewAuthorizationError("Insufficient permissions")
+	}
+
+	return self.raftServer.ChangeDbUserPermissions(db, username, readPermissions, writePermissions)
+}
+
 func (self *CoordinatorImpl) SetDbAdmin(requester common.User, db, username string, isAdmin bool) error {
 	if !requester.IsClusterAdmin() && !requester.IsDbAdmin(db) {
 		return common.NewAuthorizationError("Insufficient permissions")

src/coordinator/interface.go

@@ -37,6 +37,7 @@ type ClusterConsensus interface {
 	SaveClusterAdminUser(u *cluster.ClusterAdmin) error
 	SaveDbUser(user *cluster.DbUser) error
 	ChangeDbUserPassword(db, username string, hash []byte) error
+	ChangeDbUserPermissions(db, username, readPermissions, writePermissions string) error
 
 	// an insert index of -1 will append to the end of the ring
 	AddServer(server *cluster.ClusterServer, insertIndex int) error

src/coordinator/raft_server.go

@@ -204,6 +204,12 @@ func (s *RaftServer) ChangeDbUserPassword(db, username string, hash []byte) erro
 	return err
 }
 
+func (s *RaftServer) ChangeDbUserPermissions(db, username, readPermissions, writePermissions string) error {
+	command := NewChangeDbUserPermissionsCommand(db, username, readPermissions, writePermissions)
+	_, err := s.doOrProxyCommand(command, "change_db_user_permissions")
+	return err
+}
+
 func (s *RaftServer) SaveClusterAdminUser(u *cluster.ClusterAdmin) error {
 	command := NewSaveClusterAdminCommand(u)
 	_, err := s.doOrProxyCommand(command, "save_cluster_admin_user")

src/integration/single_server_test.go

@@ -104,6 +104,45 @@ func (self *SingleServerSuite) TestSslOnly(c *C) {
 
 }
 
+func (self *SingleServerSuite) TestUserPermissions(c *C) {
+	client, err := influxdb.NewClient(&influxdb.ClientConfig{})
+	c.Assert(err, IsNil)
+	c.Assert(client.CreateDatabaseUser("db1", "limited_user", "pass", "test_should_read", ".*"), IsNil)
+
+	data := `
+[
+  {
+    "points": [
+        [1]
+    ],
+    "name": "test_should_read",
+    "columns": ["value"]
+  },
+  {
+    "points": [
+        [2]
+    ],
+    "name": "test_should_not_read",
+    "columns": ["value"]
+  }
+]`
+	self.server.WriteData(data, c)
+
+	series := self.server.RunQueryAsUser("select value from test_should_read", "s", "limited_user", "pass", true, c)
+	c.Assert(series[0].Points, HasLen, 1)
+	c.Assert(series[0].Points[0][2], Equals, float64(1))
+
+	_ = self.server.RunQueryAsUser("select value from test_should_not_read", "s", "limited_user", "pass", false, c)
+	series = self.server.RunQueryAsUser("select value from /.*/", "s", "limited_user", "pass", true, c)
+	c.Assert(series, HasLen, 1)
+	c.Assert(series[0].Name, Equals, "test_should_read")
+
+	client.UpdateDatabaseUserPermissions("db1", "limited_user", ".*", ".*")
+	self.server.WaitForServerToSync()
+	series = self.server.RunQueryAsUser("select value from /.*/", "s", "limited_user", "pass", true, c)
+	c.Assert(series, HasLen, 2)
+}
+
 // Reported by Alex in the following thread
 // https://groups.google.com/forum/#!msg/influxdb/I_Ns6xYiMOc/XilTv6BDgHgJ
 func (self *SingleServerSuite) TestAdminPermissionToDeleteData(c *C) {