Merge pull request #1596 from kmlebedev/store_s3cred

S3 credentials store in filer

weed/s3api/auth_credentials.go

@@ -3,11 +3,15 @@ package s3api
 import (
 	"bytes"
 	"fmt"
+	"github.com/chrislusf/seaweedfs/weed/pb"
+	"github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
+	"google.golang.org/grpc"
 	"io/ioutil"
 	"net/http"
 
 	xhttp "github.com/chrislusf/seaweedfs/weed/s3api/http"
 	"github.com/chrislusf/seaweedfs/weed/s3api/s3err"
+	"github.com/chrislusf/seaweedfs/weed/s3iam"
 	"github.com/golang/protobuf/jsonpb"
 
 	"github.com/chrislusf/seaweedfs/weed/glog"
@@ -44,23 +48,38 @@ type Credential struct {
 	SecretKey string
 }
 
-func NewIdentityAccessManagement(fileName string, domain string) *IdentityAccessManagement {
+func NewIdentityAccessManagement(option *S3ApiServerOption) *IdentityAccessManagement {
 	iam := &IdentityAccessManagement{
-		domain: domain,
+		domain: option.DomainName,
 	}
-	if fileName == "" {
-		return iam
+	if err := iam.loadS3ApiConfigurationFromFiler(option); err != nil {
+		glog.Warningf("fail to load config %v", err)
+	}
+	if len(iam.identities) == 0 && option.Config != "" {
+		if err := iam.loadS3ApiConfigurationFromFile(option.Config); err != nil {
+			glog.Fatalf("fail to load config file %s: %v", option.Config, err)
 		}
-	if err := iam.loadS3ApiConfiguration(fileName); err != nil {
-		glog.Fatalf("fail to load config file %s: %v", fileName, err)
 	}
 	return iam
 }
 
-func (iam *IdentityAccessManagement) loadS3ApiConfiguration(fileName string) error {
-
+func (iam *IdentityAccessManagement) loadS3ApiConfigurationFromFiler(option *S3ApiServerOption) error {
 	s3ApiConfiguration := &iam_pb.S3ApiConfiguration{}
+	return pb.WithCachedGrpcClient(func(grpcConnection *grpc.ClientConn) error {
+		client := filer_pb.NewSeaweedFilerClient(grpcConnection)
+		store := s3iam.NewIAMFilerStore(&client)
+		if err := store.LoadIAMConfig(s3ApiConfiguration); err != nil {
+			return nil
+		}
+		if err := iam.loadS3ApiConfiguration(s3ApiConfiguration); err != nil {
+			return err
+		}
+		return nil
+	}, option.FilerGrpcAddress, option.GrpcDialOption)
+}
 
+func (iam *IdentityAccessManagement) loadS3ApiConfigurationFromFile(fileName string) error {
+	s3ApiConfiguration := &iam_pb.S3ApiConfiguration{}
 	rawData, readErr := ioutil.ReadFile(fileName)
 	if readErr != nil {
 		glog.Warningf("fail to read %s : %v", fileName, readErr)
@@ -72,8 +91,14 @@ func (iam *IdentityAccessManagement) loadS3ApiConfiguration(fileName string) err
 		glog.Warningf("unmarshal error: %v", err)
 		return fmt.Errorf("unmarshal %s error: %v", fileName, err)
 	}
+	if err := iam.loadS3ApiConfiguration(s3ApiConfiguration); err != nil {
+		return err
+	}
+	return nil
+}
 
-	for _, ident := range s3ApiConfiguration.Identities {
+func (iam *IdentityAccessManagement) loadS3ApiConfiguration(config *iam_pb.S3ApiConfiguration) error {
+	for _, ident := range config.Identities {
 		t := &Identity{
 			Name:        ident.Name,
 			Credentials: nil,
@@ -90,7 +115,6 @@ func (iam *IdentityAccessManagement) loadS3ApiConfiguration(fileName string) err
 		}
 		iam.identities = append(iam.identities, t)
 	}
-
 	return nil
 }
 

weed/s3api/auto_signature_v4_test.go

@@ -57,7 +57,8 @@ func TestIsRequestPresignedSignatureV4(t *testing.T) {
 
 // Tests is requested authenticated function, tests replies for s3 errors.
 func TestIsReqAuthenticated(t *testing.T) {
-	iam := NewIdentityAccessManagement("", "")
+	option := S3ApiServerOption{}
+	iam := NewIdentityAccessManagement(&option)
 	iam.identities = []*Identity{
 		{
 			Name: "someone",
@@ -92,7 +93,8 @@ func TestIsReqAuthenticated(t *testing.T) {
 }
 
 func TestCheckAdminRequestAuthType(t *testing.T) {
-	iam := NewIdentityAccessManagement("", "")
+	option := S3ApiServerOption{}
+	iam := NewIdentityAccessManagement(&option)
 	iam.identities = []*Identity{
 		{
 			Name: "someone",

weed/s3api/filer_util.go

@@ -3,11 +3,10 @@ package s3api
 import (
 	"context"
 	"fmt"
-	"strings"
-
 	"github.com/chrislusf/seaweedfs/weed/glog"
 	"github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
 	"github.com/chrislusf/seaweedfs/weed/util"
+	"strings"
 )
 
 func (s3a *S3ApiServer) mkdir(parentDirectoryPath string, dirName string, fn func(entry *filer_pb.Entry)) error {

weed/s3api/s3api_server.go

@@ -27,7 +27,7 @@ type S3ApiServer struct {
 func NewS3ApiServer(router *mux.Router, option *S3ApiServerOption) (s3ApiServer *S3ApiServer, err error) {
 	s3ApiServer = &S3ApiServer{
 		option: option,
-		iam:    NewIdentityAccessManagement(option.Config, option.DomainName),
+		iam:    NewIdentityAccessManagement(option),
 	}
 
 	s3ApiServer.registerRouter(router)

weed/s3iam/s3iam_filer_store.go

@@ -0,0 +1,95 @@
+package s3iam
+
+import (
+	"github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
+	"github.com/chrislusf/seaweedfs/weed/pb/iam_pb"
+	"time"
+
+	proto "github.com/golang/protobuf/proto"
+)
+
+const (
+	iamConfigPrefix = "/etc/iam"
+	iamIdentityFile = "identity.json"
+)
+
+type IAMFilerStore struct {
+	client *filer_pb.SeaweedFilerClient
+}
+
+func NewIAMFilerStore(client *filer_pb.SeaweedFilerClient) *IAMFilerStore {
+	return &IAMFilerStore{client: client}
+}
+
+func (ifs *IAMFilerStore) getIAMConfigRequest() *filer_pb.LookupDirectoryEntryRequest {
+	return &filer_pb.LookupDirectoryEntryRequest{
+		Directory: iamConfigPrefix,
+		Name:      iamIdentityFile,
+	}
+}
+
+func (ifs *IAMFilerStore) LoadIAMConfig(config *iam_pb.S3ApiConfiguration) error {
+	resp, err := filer_pb.LookupEntry(*ifs.client, ifs.getIAMConfigRequest())
+	if err != nil {
+		return err
+	}
+	err = ifs.loadIAMConfigFromEntry(resp.Entry, config)
+	if err != nil {
+		return err
+	}
+	return nil
+}
+
+func (ifs *IAMFilerStore) SaveIAMConfig(config *iam_pb.S3ApiConfiguration) error {
+	entry := &filer_pb.Entry{
+		Name:        iamIdentityFile,
+		IsDirectory: false,
+		Attributes: &filer_pb.FuseAttributes{
+			Mtime:       time.Now().Unix(),
+			Crtime:      time.Now().Unix(),
+			FileMode:    uint32(0644),
+			Collection:  "",
+			Replication: "",
+		},
+		Content: []byte{},
+	}
+	err := ifs.saveIAMConfigToEntry(entry, config)
+	if err != nil {
+		return err
+	}
+	_, err = filer_pb.LookupEntry(*ifs.client, ifs.getIAMConfigRequest())
+	if err == filer_pb.ErrNotFound {
+		err = filer_pb.CreateEntry(*ifs.client, &filer_pb.CreateEntryRequest{
+			Directory:          iamConfigPrefix,
+			Entry:              entry,
+			IsFromOtherCluster: false,
+			Signatures:         nil,
+		})
+	} else {
+		err = filer_pb.UpdateEntry(*ifs.client, &filer_pb.UpdateEntryRequest{
+			Directory:          iamConfigPrefix,
+			Entry:              entry,
+			IsFromOtherCluster: false,
+			Signatures:         nil,
+		})
+	}
+	if err != nil {
+		return err
+	}
+	return nil
+}
+
+func (ifs *IAMFilerStore) loadIAMConfigFromEntry(entry *filer_pb.Entry, config *iam_pb.S3ApiConfiguration) error {
+	if err := proto.Unmarshal(entry.Content, config); err != nil {
+		return err
+	}
+	return nil
+}
+
+func (ifs *IAMFilerStore) saveIAMConfigToEntry(entry *filer_pb.Entry, config *iam_pb.S3ApiConfiguration) (err error) {
+	entry.Content, err = proto.Marshal(config)
+	if err != nil {
+		return err
+	}
+	return nil
+}

weed/s3iam/s3iam_filer_store_test.go

@@ -0,0 +1,65 @@
+package s3iam
+
+import (
+	"testing"
+
+	"github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
+	"github.com/chrislusf/seaweedfs/weed/pb/iam_pb"
+
+	"github.com/stretchr/testify/assert"
+)
+
+const (
+	ACTION_READ    = "Read"
+	ACTION_WRITE   = "Write"
+	ACTION_ADMIN   = "Admin"
+	ACTION_TAGGING = "Tagging"
+	ACTION_LIST    = "List"
+)
+
+func TestS3Conf(t *testing.T) {
+	ifs := &IAMFilerStore{}
+	s3Conf := &iam_pb.S3ApiConfiguration{
+		Identities: []*iam_pb.Identity{
+			{
+				Name: "some_name",
+				Credentials: []*iam_pb.Credential{
+					{
+						AccessKey: "some_access_key1",
+						SecretKey: "some_secret_key1",
+					},
+				},
+				Actions: []string{
+					ACTION_ADMIN,
+					ACTION_READ,
+					ACTION_WRITE,
+				},
+			},
+			{
+				Name: "some_read_only_user",
+				Credentials: []*iam_pb.Credential{
+					{
+						AccessKey: "some_access_key2",
+						SecretKey: "some_secret_key2",
+					},
+				},
+				Actions: []string{
+					ACTION_READ,
+					ACTION_TAGGING,
+					ACTION_LIST,
+				},
+			},
+		},
+	}
+	entry := filer_pb.Entry{}
+	err := ifs.saveIAMConfigToEntry(&entry, s3Conf)
+	assert.Equal(t, err, nil)
+	s3ConfSaved := &iam_pb.S3ApiConfiguration{}
+	err = ifs.loadIAMConfigFromEntry(&entry, s3ConfSaved)
+	assert.Equal(t, err, nil)
+
+	assert.Equal(t, "some_name", s3ConfSaved.Identities[0].Name)
+	assert.Equal(t, "some_read_only_user", s3ConfSaved.Identities[1].Name)
+	assert.Equal(t, "some_access_key1", s3ConfSaved.Identities[0].Credentials[0].AccessKey)
+	assert.Equal(t, "some_secret_key2", s3ConfSaved.Identities[1].Credentials[0].SecretKey)
+}

weed/shell/command_s3_configure.go

@@ -0,0 +1,174 @@
+package shell
+
+import (
+	"flag"
+	"fmt"
+	"io"
+	"sort"
+	"strings"
+
+	"github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
+	"github.com/chrislusf/seaweedfs/weed/pb/iam_pb"
+	"github.com/chrislusf/seaweedfs/weed/s3iam"
+)
+
+func init() {
+	Commands = append(Commands, &commandS3Configure{})
+}
+
+type commandS3Configure struct {
+}
+
+func (c *commandS3Configure) Name() string {
+	return "s3.configure"
+}
+
+func (c *commandS3Configure) Help() string {
+	return `configure and apply s3 options for each bucket
+	# see the current configuration file content
+	s3.configure
+	`
+}
+
+func (c *commandS3Configure) Do(args []string, commandEnv *CommandEnv, writer io.Writer) (err error) {
+	s3ConfigureCommand := flag.NewFlagSet(c.Name(), flag.ContinueOnError)
+	actions := s3ConfigureCommand.String("actions", "", "actions names")
+	user := s3ConfigureCommand.String("user", "", "user name")
+	buckets := s3ConfigureCommand.String("buckets", "", "bucket name")
+	accessKey := s3ConfigureCommand.String("access_key", "", "specify the access key")
+	secretKey := s3ConfigureCommand.String("secret_key", "", "specify the secret key")
+	isDelete := s3ConfigureCommand.Bool("delete", false, "delete users, actions or access keys")
+	apply := s3ConfigureCommand.Bool("apply", false, "update and apply s3 configuration")
+
+	if err = s3ConfigureCommand.Parse(args); err != nil {
+		return nil
+	}
+
+	s3cfg := &iam_pb.S3ApiConfiguration{}
+	ifs := &s3iam.IAMFilerStore{}
+	if err = commandEnv.WithFilerClient(func(client filer_pb.SeaweedFilerClient) error {
+		ifs = s3iam.NewIAMFilerStore(&client)
+		if err := ifs.LoadIAMConfig(s3cfg); err != nil {
+			return nil
+		}
+		return nil
+	}); err != nil {
+		return err
+	}
+
+	idx := 0
+	changed := false
+	if *user != "" {
+		for i, identity := range s3cfg.Identities {
+			if *user == identity.Name {
+				idx = i
+				changed = true
+				break
+			}
+		}
+	}
+	var cmdActions []string
+	for _, action := range strings.Split(*actions, ",") {
+		if *buckets == "" {
+			cmdActions = append(cmdActions, action)
+		} else {
+			for _, bucket := range strings.Split(*buckets, ",") {
+				cmdActions = append(cmdActions, fmt.Sprintf("%s:%s", action, bucket))
+			}
+		}
+	}
+	if changed {
+		if *isDelete {
+			var exists []int
+			for _, cmdAction := range cmdActions {
+				for i, currentAction := range s3cfg.Identities[idx].Actions {
+					if cmdAction == currentAction {
+						exists = append(exists, i)
+					}
+				}
+			}
+			sort.Sort(sort.Reverse(sort.IntSlice(exists)))
+			for _, i := range exists {
+				s3cfg.Identities[idx].Actions = append(
+					s3cfg.Identities[idx].Actions[:i],
+					s3cfg.Identities[idx].Actions[i+1:]...,
+				)
+			}
+			if *accessKey != "" {
+				exists = []int{}
+				for i, credential := range s3cfg.Identities[idx].Credentials {
+					if credential.AccessKey == *accessKey {
+						exists = append(exists, i)
+					}
+				}
+				sort.Sort(sort.Reverse(sort.IntSlice(exists)))
+				for _, i := range exists {
+					s3cfg.Identities[idx].Credentials = append(
+						s3cfg.Identities[idx].Credentials[:i],
+						s3cfg.Identities[idx].Credentials[:i+1]...,
+					)
+				}
+
+			}
+			if *actions == "" && *accessKey == "" && *buckets == "" {
+				s3cfg.Identities = append(s3cfg.Identities[:idx], s3cfg.Identities[idx+1:]...)
+			}
+		} else {
+			if *actions != "" {
+				for _, cmdAction := range cmdActions {
+					found := false
+					for _, action := range s3cfg.Identities[idx].Actions {
+						if cmdAction == action {
+							found = true
+							break
+						}
+					}
+					if !found {
+						s3cfg.Identities[idx].Actions = append(s3cfg.Identities[idx].Actions, cmdAction)
+					}
+				}
+			}
+			if *accessKey != "" && *user != "anonymous" {
+				found := false
+				for _, credential := range s3cfg.Identities[idx].Credentials {
+					if credential.AccessKey == *accessKey {
+						found = true
+						credential.SecretKey = *secretKey
+						break
+					}
+				}
+				if !found {
+					s3cfg.Identities[idx].Credentials = append(s3cfg.Identities[idx].Credentials, &iam_pb.Credential{
+						AccessKey: *accessKey,
+						SecretKey: *secretKey,
+					})
+				}
+			}
+		}
+	} else if *user != "" && *actions != "" {
+		identity := iam_pb.Identity{
+			Name:        *user,
+			Actions:     cmdActions,
+			Credentials: []*iam_pb.Credential{},
+		}
+		if *user != "anonymous" {
+			identity.Credentials = append(identity.Credentials,
+				&iam_pb.Credential{AccessKey: *accessKey, SecretKey: *secretKey})
+		}
+		s3cfg.Identities = append(s3cfg.Identities, &identity)
+	}
+
+	for _, identity := range s3cfg.Identities {
+		fmt.Fprintf(writer, fmt.Sprintf("%+v\n", identity))
+	}
+
+	fmt.Fprintln(writer)
+
+	if *apply {
+		if err := ifs.SaveIAMConfig(s3cfg); err != nil {
+			return err
+		}
+	}
+
+	return nil
+}