hadoop: support kerberos

.github/workflows/sdktest.yml

@@ -79,13 +79,16 @@ jobs:
 
       - name: Sdk Test
         run: |
+          sudo sh sdk/java/kerberos.sh
           make -C sdk/java/libjfs
           cd sdk/java
-          sudo mvn test -B -Dtest=\!io.juicefs.permission.**
+          sudo mvn test -B -Dtest=\!io.juicefs.permission.**,\!io.juicefs.kerberos.**
           sudo mvn test -B -Dflink.version=1.17.2 -Dtest=io.juicefs.JuiceFileSystemTest#testFlinkHadoopRecoverableWriter
           # ranger test
           sudo JUICEFS_RANGER_TEST=1 mvn test -B -Dtest=io.juicefs.permission.RangerPermissionCheckerTest,\!io.juicefs.permission.RangerPermissionCheckerTest#testRangerCheckerInitFailed
           sudo mvn test -B -Dtest=io.juicefs.permission.RangerPermissionCheckerTest#testRangerCheckerInitFailed
+          # kerberos test
+          sudo mvn test -B -Dtest=io.juicefs.kerberos.KerberosTest
 
           sudo mvn package -B -Dmaven.test.skip=true --quiet -Dmaven.javadoc.skip=true
           expect=$(git rev-parse --short HEAD | cut -b 1-7)

cmd/config.go

@@ -281,6 +281,11 @@ func config(ctx *cli.Context) error {
 				format.MinClientVersion = "1.3.0-A"
 				clientVer = true
 			}
+		case "kerberos-config-file":
+			msg.WriteString(fmt.Sprintf("%s: updated\n", flag))
+			format.KerbConf = readKerbConf(ctx.String(flag))
+			format.MinClientVersion = "1.4.0-A"
+			clientVer = true
 		}
 	}
 	if msg.Len() == 0 {

cmd/format.go

@@ -205,6 +205,10 @@ func formatManagementFlags() []cli.Flag {
 			Name:  "ranger-service",
 			Usage: "Name of the Ranger service used For JuiceFS",
 		},
+		&cli.StringFlag{
+			Name:  "kerberos-config-file",
+			Usage: "Path to Kerberos configuration file",
+		},
 	})
 }
 
@@ -393,6 +397,17 @@ func loadEncrypt(keyPath string) string {
 	return string(pem)
 }
 
+func readKerbConf(file string) string {
+	if file == "" {
+		return ""
+	}
+	data, err := os.ReadFile(file)
+	if err != nil {
+		logger.Fatalf("load Kerberos config from %s: %s", file, err)
+	}
+	return string(data)
+}
+
 func format(c *cli.Context) error {
 	setup(c, 2)
 	removePassword(c.Args().Get(0))
@@ -459,6 +474,8 @@ func format(c *cli.Context) error {
 				format.RangerRestUrl = c.String(flag)
 			case "ranger-service":
 				format.RangerService = c.String(flag)
+			case "kerberos-config-file":
+				format.KerbConf = readKerbConf(c.String(flag))
 			}
 		}
 	} else if strings.HasPrefix(err.Error(), "database is not formatted") {
@@ -488,6 +505,7 @@ func format(c *cli.Context) error {
 			EnableACL:        c.Bool("enable-acl"),
 			RangerRestUrl:    c.String("ranger-rest-url"),
 			RangerService:    c.String("ranger-service"),
+			KerbConf:         readKerbConf(c.String("kerberos-config-file")),
 		}
 		if format.EnableACL {
 			format.MinClientVersion = "1.2.0-A"

pkg/fs/fs.go

@@ -139,6 +139,10 @@ type FileSystem struct {
 	store       chunk.ChunkStore
 	cacheFiller *vfs.CacheFiller
 
+	Superuser          string
+	Supergroup         string
+	OnSuperuserChanged []func()
+
 	cacheM          sync.Mutex
 	entries         map[Ino]map[string]*entryCache
 	attrs           map[Ino]*attrCache

pkg/meta/base.go

@@ -144,6 +144,13 @@ type engine interface {
 	doGetFacl(ctx Context, ino Ino, aclType uint8, aclId uint32, rule *aclAPI.Rule) syscall.Errno
 	cacheACLs(ctx Context) error
 
+	// kerberos delegation token
+	doStoreToken(ctx Context, token []byte) (id uint32, st syscall.Errno)
+	doUpdateToken(ctx Context, id uint32, token []byte) syscall.Errno
+	doLoadToken(ctx Context, id uint32) (token []byte, st syscall.Errno)
+	doDeleteTokens(ctx Context, ids []uint32) syscall.Errno
+	doListTokens(ctx Context) (tokens map[uint32][]byte, st syscall.Errno)
+
 	newDirHandler(inode Ino, plus bool, entries []*Entry) DirHandler
 
 	dump(ctx Context, opt *DumpOption, ch chan<- *dumpedResult) error
@@ -3154,6 +3161,31 @@ func (m *baseMeta) GetFacl(ctx Context, ino Ino, aclType uint8, rule *aclAPI.Rul
 	return m.en.doGetFacl(ctx, ino, aclType, aclAPI.None, rule)
 }
 
+func (m *baseMeta) StoreToken(ctx Context, token []byte) (id uint32, st syscall.Errno) {
+	defer m.timeit("StoreToken", time.Now())
+	return m.en.doStoreToken(ctx, token)
+}
+
+func (m *baseMeta) UpdateToken(ctx Context, id uint32, token []byte) syscall.Errno {
+	defer m.timeit("UpdateToken", time.Now())
+	return m.en.doUpdateToken(ctx, id, token)
+}
+
+func (m *baseMeta) LoadToken(ctx Context, id uint32) (token []byte, st syscall.Errno) {
+	defer m.timeit("LoadToken", time.Now())
+	return m.en.doLoadToken(ctx, id)
+}
+
+func (m *baseMeta) DeleteTokens(ctx Context, ids []uint32) syscall.Errno {
+	defer m.timeit("DeleteTokens", time.Now())
+	return m.en.doDeleteTokens(ctx, ids)
+}
+
+func (m *baseMeta) ListTokens(ctx Context) (tokens map[uint32][]byte, st syscall.Errno) {
+	defer m.timeit("ListTokens", time.Now())
+	return m.en.doListTokens(ctx)
+}
+
 func inGroup(ctx Context, gid uint32) bool {
 	for _, egid := range ctx.Gids() {
 		if egid == gid {

pkg/meta/config.go

@@ -100,6 +100,9 @@ type Format struct {
 	EnableACL        bool
 	RangerRestUrl    string `json:",omitempty"`
 	RangerService    string `json:",omitempty"`
+
+	//kerberos
+	KerbConf string `json:",omitempty"`
 }
 
 func (f *Format) update(old *Format, force bool) error {

pkg/meta/interface.go

@@ -529,6 +529,13 @@ type Meta interface {
 
 	SetFacl(ctx Context, ino Ino, aclType uint8, n *aclAPI.Rule) syscall.Errno
 	GetFacl(ctx Context, ino Ino, aclType uint8, n *aclAPI.Rule) syscall.Errno
+
+	// kerberos
+	StoreToken(ctx Context, token []byte) (id uint32, st syscall.Errno)
+	UpdateToken(ctx Context, id uint32, token []byte) syscall.Errno
+	LoadToken(ctx Context, id uint32) (token []byte, st syscall.Errno)
+	DeleteTokens(ctx Context, ids []uint32) syscall.Errno
+	ListTokens(ctx Context) (tokens map[uint32][]byte, st syscall.Errno)
 }
 
 type Creator func(driver, addr string, conf *Config) (Meta, error)

pkg/meta/redis.go

@@ -77,6 +77,7 @@ import (
 	Quota used space:  dirQuotaUsedSpace -> { $inode -> usedSpace }
 	Quota used inodes: dirQuotaUsedInodes -> { $inode -> usedInodes }
 	Acl: acl -> { $acl_id -> acl }
+	KrbToken: krbToken -> { $token_id -> token }
 
 	Redis features:
 	  Sorted Set: 1.2+
@@ -655,6 +656,10 @@ func (m *redisMeta) aclKey() string {
 	return m.prefix + "acl"
 }
 
+func (m *redisMeta) krbTokenKey() string {
+	return m.prefix + "krbToken"
+}
+
 func (m *redisMeta) delfiles() string {
 	return m.prefix + "delfiles"
 }
@@ -1340,7 +1345,7 @@ func (m *redisMeta) doMknod(ctx Context, parent Ino, name string, _type uint8, m
 		if (pattr.Flags & FlagImmutable) != 0 {
 			return syscall.EPERM
 		}
-		if (pattr.Flags&FlagSkipTrash) != 0 {
+		if (pattr.Flags & FlagSkipTrash) != 0 {
 			attr.Flags |= FlagSkipTrash
 		}
 
@@ -1694,7 +1699,7 @@ func (m *redisMeta) doRmdir(ctx Context, parent Ino, name string, pinode *Ino, o
 			if ctx.Uid() != 0 && pattr.Mode&01000 != 0 && ctx.Uid() != pattr.Uid && ctx.Uid() != attr.Uid {
 				return syscall.EACCES
 			}
-			if (attr.Flags&FlagSkipTrash) != 0 {
+			if (attr.Flags & FlagSkipTrash) != 0 {
 				trash = 0
 			}
 			if trash > 0 {
@@ -1873,7 +1878,7 @@ func (m *redisMeta) doRename(ctx Context, parentSrc Ino, nameSrc string, parentD
 			if (tattr.Flags&FlagAppend) != 0 || (tattr.Flags&FlagImmutable) != 0 {
 				return syscall.EPERM
 			}
-			if (tattr.Flags&FlagSkipTrash) != 0 {
+			if (tattr.Flags & FlagSkipTrash) != 0 {
 				trash = 0
 			}
 			tattr.Ctime = now.Unix()
@@ -4759,6 +4764,79 @@ func (m *redisMeta) loadDumpedACLs(ctx Context) error {
 	}, m.inodeKey(RootInode))
 }
 
+func (m *redisMeta) doStoreToken(ctx Context, token []byte) (id uint32, st syscall.Errno) {
+	err := m.txn(ctx, func(tx *redis.Tx) error {
+		newId, err := m.incrCounter(krbTokenCounter, 1)
+		if err != nil {
+			return err
+		}
+		err = tx.HSet(ctx, m.krbTokenKey(), strconv.FormatUint(uint64(newId), 10), token).Err()
+		if err == nil {
+			id = uint32(newId)
+		}
+		return err
+	}, m.krbTokenKey())
+	return id, errno(err)
+}
+
+func (m *redisMeta) doUpdateToken(ctx Context, id uint32, token []byte) syscall.Errno {
+	return errno(m.txn(ctx, func(tx *redis.Tx) error {
+		exist, err := tx.HExists(ctx, m.krbTokenKey(), strconv.FormatUint(uint64(id), 10)).Result()
+		if err != nil {
+			return err
+		}
+		if !exist {
+			return syscall.ENOENT
+		}
+		return tx.HSet(ctx, m.krbTokenKey(), strconv.FormatUint(uint64(id), 10), token).Err()
+	}, m.krbTokenKey()))
+}
+
+func (m *redisMeta) doLoadToken(ctx Context, id uint32) (token []byte, st syscall.Errno) {
+	err := m.txn(ctx, func(tx *redis.Tx) error {
+		val, err := tx.HGet(ctx, m.krbTokenKey(), strconv.FormatUint(uint64(id), 10)).Bytes()
+		if err != nil {
+			return err
+		}
+		if val == nil {
+			return syscall.ENOENT
+		}
+		token = val
+		return nil
+	}, m.krbTokenKey())
+	return token, errno(err)
+}
+
+func (m *redisMeta) doDeleteTokens(ctx Context, ids []uint32) syscall.Errno {
+	return errno(m.txn(ctx, func(tx *redis.Tx) error {
+		strIds := make([]string, len(ids))
+		for i, id := range ids {
+			strIds[i] = strconv.FormatUint(uint64(id), 10)
+		}
+		return tx.HDel(ctx, m.krbTokenKey(), strIds...).Err()
+	}, m.krbTokenKey()))
+}
+
+func (m *redisMeta) doListTokens(ctx Context) (tokens map[uint32][]byte, st syscall.Errno) {
+	tokens = make(map[uint32][]byte)
+	err := m.txn(ctx, func(tx *redis.Tx) error {
+		vals, err := tx.HGetAll(ctx, m.krbTokenKey()).Result()
+		if err != nil {
+			return err
+		}
+		for k, v := range vals {
+			id, err := strconv.ParseUint(k, 10, 32)
+			if err != nil {
+				logger.Errorf("parse token id: %s: %v", k, err)
+				continue
+			}
+			tokens[uint32(id)] = []byte(v)
+		}
+		return nil
+	}, m.krbTokenKey())
+	return tokens, errno(err)
+}
+
 func (m *redisMeta) newDirHandler(inode Ino, plus bool, entries []*Entry) DirHandler {
 	return &redisDirHandler{
 		en:          m,