Add CreateUser flow.

errors.go

@@ -15,12 +15,26 @@ var (
 	// ErrPathRequired is returned when opening a server without a path.
 	ErrPathRequired = errors.New("path required")
 
-	// ErrDatabaseExists is returned when creating a database with the same
-	// name as an existing database.
+	// ErrDatabaseExists is returned when creating a duplicate database.
 	ErrDatabaseExists = errors.New("database exists")
 
 	// ErrDatabaseNotFound is returned when dropping a non-existent database.
 	ErrDatabaseNotFound = errors.New("database not found")
+
+	// ErrDatabaseRequired is returned when using a blank database name.
+	ErrDatabaseRequired = errors.New("database required")
+
+	// ErrUserExists is returned when creating a duplicate user.
+	ErrUserExists = errors.New("user exists")
+
+	// ErrUserNotFound is returned when deleting a non-existent user.
+	ErrUserNotFound = errors.New("user not found")
+
+	// ErrUsernameRequired is returned when using a blank username.
+	ErrUsernameRequired = errors.New("username required")
+
+	// ErrInvalidUsername is returned when using a username with invalid characters.
+	ErrInvalidUsername = errors.New("invalid username")
 )
 
 const (

server.go

@@ -167,7 +167,9 @@ func (s *Server) applyCreateDatabase(m *messaging.Message) error {
 	}
 
 	// Create database entry.
-	s.databases[c.Name] = &Database{name: c.Name}
+	db := newDatabase(s)
+	db.name = c.Name
+	s.databases[c.Name] = db
 	return nil
 }
 
@@ -200,6 +202,65 @@ type deleteDatabaseCommand struct {
 	Name string `json:"name"`
 }
 
+func (s *Server) applyCreateDBUser(m *messaging.Message) error {
+	var c createDBUserCommand
+	mustUnmarshal(m.Data, &c)
+
+	s.mu.Lock()
+	defer s.mu.Unlock()
+
+	// Validate user.
+	if c.Username == "" {
+		return ErrUsernameRequired
+	} else if !isValidName(c.Username) {
+		return ErrInvalidUsername
+	} else if c.Database == "" {
+		return ErrDatabaseRequired
+	}
+
+	// Retrieve the database.
+	db := s.databases[c.Database]
+	if s.databases[c.Database] == nil {
+		return ErrDatabaseNotFound
+	} else if db.User(c.Username) != nil {
+		return ErrUserExists
+	}
+
+	// Generate the hash of the password.
+	hash, err := HashPassword(c.Password)
+	if err != nil {
+		return err
+	}
+
+	// Setup matchers.
+	rmatcher := []*Matcher{{true, ".*"}}
+	wmatcher := []*Matcher{{true, ".*"}}
+	if len(c.Permissions) == 2 {
+		rmatcher[0].Name = c.Permissions[0]
+		wmatcher[0].Name = c.Permissions[1]
+	}
+
+	// Create the user.
+	u := &DBUser{
+		CommonUser: CommonUser{
+			Name: c.Username,
+			Hash: string(hash),
+		},
+		DB:       c.Database,
+		ReadFrom: rmatcher,
+		WriteTo:  wmatcher,
+		IsAdmin:  false,
+	}
+	return db.saveUser(u)
+}
+
+type createDBUserCommand struct {
+	Database    string   `json:"database"`
+	Username    string   `json:"username"`
+	Password    string   `json:"password"`
+	Permissions []string `json:"permissions"`
+}
+
 // WriteSeries writes series data to the broker.
 func (s *Server) WriteSeries(u *ClusterAdmin, database string, series *protocol.Series) error {
 	// TODO:
@@ -225,6 +286,8 @@ func (s *Server) processor(done chan struct{}) {
 			err = s.applyCreateDatabase(m)
 		case deleteDatabaseMessageType:
 			err = s.applyDeleteDatabase(m)
+		case createDBUserMessageType:
+			err = s.applyCreateDBUser(m)
 		}
 
 		// Sync high water mark and errors for broadcast topic.
@@ -250,9 +313,20 @@ type MessagingClient interface {
 
 // Database represents a collection of shard spaces.
 type Database struct {
-	mu          sync.RWMutex
-	name        string
-	shardSpaces map[string]*ShardSpace
+	mu     sync.RWMutex
+	server *Server
+	name   string
+	users  map[string]*DBUser
+	spaces map[string]*ShardSpace
+}
+
+// newDatabase returns an instance of Database associated with a server.
+func newDatabase(s *Server) *Database {
+	return &Database{
+		server: s,
+		users:  make(map[string]*DBUser),
+		spaces: make(map[string]*ShardSpace),
+	}
 }
 
 // Name returns the database name.
@@ -262,6 +336,35 @@ func (db *Database) Name() string {
 	return db.name
 }
 
+// User returns a database user by name.
+func (db *Database) User(name string) *DBUser {
+	db.mu.Lock()
+	defer db.mu.Unlock()
+	return db.users[name]
+}
+
+// CreateUser creates a user in the database.
+func (db *Database) CreateUser(username, password string, permissions []string) error {
+	// TODO: Authorization.
+
+	c := &createDBUserCommand{
+		Database:    db.Name(),
+		Username:    username,
+		Password:    password,
+		Permissions: permissions,
+	}
+	_, err := db.server.broadcast(createDBUserMessageType, c)
+	return err
+}
+
+// saveUser persists a user to the database.
+func (db *Database) saveUser(u *DBUser) error {
+	db.mu.Lock()
+	db.users[u.Name] = u
+	db.mu.Unlock()
+	return nil
+}
+
 // ShardSpace represents a policy for creating new shards in a database.
 type ShardSpace struct {
 	// Unique name within database. Required.

server_test.go

@@ -4,6 +4,7 @@ import (
 	"fmt"
 	"io/ioutil"
 	"os"
+	"reflect"
 	"testing"
 
 	"github.com/influxdb/influxdb"
@@ -86,6 +87,30 @@ func TestServer_DropDatabase_ErrDatabaseNotFound(t *testing.T) {
 	}
 }
 
+// Ensure the server can create a new user.
+func TestServer_CreateUser(t *testing.T) {
+	s := OpenServer(NewMessagingClient())
+	defer s.Close()
+
+	// Create a database.
+	if err := s.CreateDatabase("foo"); err != nil {
+		t.Fatal(err)
+	}
+	db := s.Database("foo")
+
+	// Create a user on the database.
+	if err := db.CreateUser("susy", "pass", nil); err != nil {
+		t.Fatal(err)
+	}
+
+	// Verify that the user exists.
+	if u := db.User("susy"); u == nil {
+		t.Fatalf("user not found")
+	} else if reflect.DeepEqual(u, nil) {
+		t.Fatalf("user mismatch: %#v", u)
+	}
+}
+
 // Server is a wrapping test struct for influxdb.Server.
 type Server struct {
 	*influxdb.Server

user.go

@@ -2,6 +2,7 @@ package influxdb
 
 import (
 	"regexp"
+	"strings"
 
 	"code.google.com/p/go.crypto/bcrypt"
 	"github.com/influxdb/go-cache"
@@ -74,7 +75,7 @@ type DBUser struct {
 	IsAdmin    bool       `json:"is_admin"`
 }
 
-func (u *DBUser) IsDbAdmin(db string) bool {
+func (u *DBUser) IsDBAdmin(db string) bool {
 	return u.IsAdmin && u.DB == db
 }
 
@@ -132,3 +133,8 @@ func HashPassword(password string) ([]byte, error) {
 	// to brute force, since it will be really slow and impractical
 	return bcrypt.GenerateFromPassword([]byte(password), 10)
 }
+
+// isValidName returns true if the name contains no invalid characters.
+func isValidName(name string) bool {
+	return !strings.Contains(name, "%")
+}