redis/lua-script: pass gids to check permission

pkg/meta/base_test.go

@@ -123,6 +123,7 @@ func testMeta(t *testing.T, m Meta) {
 	testTrash(t, m)
 	testParents(t, m)
 	testRemove(t, m)
+	testResolve(t, m)
 	testStickyBit(t, m)
 	testLocks(t, m)
 	testListLocks(t, m)
@@ -979,6 +980,45 @@ func testLocks(t *testing.T, m Meta) {
 	}
 }
 
+func testResolve(t *testing.T, m Meta) {
+	var inode, parent Ino
+	var attr, pattr Attr
+	if st := m.Mkdir(NewContext(1, 65534, []uint32{65534}), 1, "d", 0770, 0, 0, &parent, &pattr); st != 0 {
+		t.Fatalf("mkdir d: %s", st)
+	}
+	if pattr.Uid != 65534 || pattr.Gid != 65534 {
+		t.Fatalf("attr %+v", pattr)
+	}
+	if st := m.Create(NewContext(1, 65534, []uint32{65534}), parent, "f", 0644, 0, 0, &inode, &attr); st != 0 {
+		t.Fatalf("create /d/f: %s", st)
+	}
+
+	defer func() {
+		if st := m.Remove(NewContext(0, 65534, []uint32{65534}), parent, "f", nil); st != 0 {
+			t.Fatalf("remove /d/f by owner: %s", st)
+		}
+		if st := m.Rmdir(NewContext(0, 65534, []uint32{65534}), 1, "d"); st != 0 {
+			t.Fatalf("rmdir /d by owner: %s", st)
+		}
+	}()
+
+	if st := m.Resolve(NewContext(0, 65534, []uint32{65534}), 1, "/d/f", &inode, &attr); st != 0 {
+		if st == syscall.ENOTSUP {
+			return
+		}
+		t.Fatalf("resolve /d/f by owner: %s", st)
+	}
+	if st := m.Resolve(NewContext(0, 65533, []uint32{65534}), 1, "/d/f", &inode, &attr); st != 0 {
+		t.Fatalf("resolve /d/f by group: %s", st)
+	}
+	if st := m.Resolve(NewContext(0, 65533, []uint32{65533, 65534}), 1, "/d/f", &inode, &attr); st != 0 {
+		t.Fatalf("resolve /d/f by multi-group: %s", st)
+	}
+	if st := m.Resolve(NewContext(0, 65533, []uint32{65533}), 1, "/d/f", &inode, &attr); st != syscall.EACCES {
+		t.Fatalf("resolve /d/f by non-group: %s", st)
+	}
+}
+
 func testRemove(t *testing.T, m Meta) {
 	ctx := Background
 	var inode, parent Ino

pkg/meta/lua_scripts.go

@@ -14,7 +14,7 @@
  * limitations under the License.
  */
 
-//nolint
+// nolint
 package meta
 
 const scriptLookup = `
@@ -55,7 +55,16 @@ local function lookup(parent, name)
     return struct.unpack(">BI8", buf)
 end
 
-local function can_access(ino, uid, gid)
+local function has_value(tab, val)
+    for index, value in ipairs(tab) do
+        if value == val then
+            return true
+        end
+    end
+    return false
+end
+
+local function can_access(ino, uid, gids)
     if uid == 0 then
         return true
     end
@@ -64,23 +73,23 @@ local function can_access(ino, uid, gid)
     local mode = 0
     if attr.uid == uid then
         mode = math.floor(attr.mode / 64) % 8
-    elseif attr.gid == gid then
+    elseif has_value(gids, tostring(attr.gid)) then
         mode = math.floor(attr.mode / 8) % 8
     else
         mode = attr.mode % 8
     end
     return mode % 2 == 1
 end
 
-local function resolve(parent, path, uid, gid)
+local function resolve(parent, path, uid, gids)
     local _maxIno = 4503599627370495
     local _type = 2
     for name in string.gmatch(path, "[^/]+") do
         if _type == 3 or parent > _maxIno then
             error("ENOTSUP")
         elseif _type ~= 2 then
             error("ENOTDIR")
-        elseif parent > 1 and not can_access(parent, uid, gid) then 
+        elseif parent > 1 and not can_access(parent, uid, gids) then 
             error("EACCESS")
         end
         _type, parent = lookup(parent, name)
@@ -91,5 +100,5 @@ local function resolve(parent, path, uid, gid)
     return {parent, redis.call('GET', "i" .. string.format("%.f", parent))}
 end
 
-return resolve(tonumber(KEYS[1]), KEYS[2], tonumber(KEYS[3]), tonumber(KEYS[4]))
+return resolve(tonumber(KEYS[1]), KEYS[2], tonumber(KEYS[3]), ARGV)
 `

pkg/meta/redis.go

@@ -777,10 +777,13 @@ func (m *redisMeta) Resolve(ctx Context, parent Ino, path string, inode *Ino, at
 	}
 	defer m.timeit("Resolve", time.Now())
 	parent = m.checkRoot(parent)
-	args := []string{parent.String(), path,
-		strconv.FormatUint(uint64(ctx.Uid()), 10),
-		strconv.FormatUint(uint64(ctx.Gid()), 10)}
-	res, err := m.rdb.EvalSha(ctx, m.shaResolve, args).Result()
+	keys := []string{parent.String(), path,
+		strconv.FormatUint(uint64(ctx.Uid()), 10)}
+	var gids []interface{}
+	for _, gid := range ctx.Gids() {
+		gids = append(gids, strconv.FormatUint(uint64(gid), 10))
+	}
+	res, err := m.rdb.EvalSha(ctx, m.shaResolve, keys, gids...).Result()
 	var returnedIno int64
 	var returnedAttr string
 	st := m.handleLuaResult("resolve", res, err, &returnedIno, &returnedAttr)

sdk/java/libjfs/guid.go

@@ -71,7 +71,10 @@ func (m *mapping) lookupUser(name string) uint32 {
 		return id
 	}
 	if !m.local {
-		return m.genGuid(name)
+		id := m.genGuid(name)
+		m.usernames[name] = id
+		m.userIDs[id] = name
+		return id
 	}
 	if name == "root" { // root in hdfs sdk is a normal user
 		id = m.genGuid(name)

sdk/java/src/test/java/io/juicefs/JuiceFileSystemTest.java

@@ -894,4 +894,30 @@ public void testInnerSymlink() throws Exception {
     assertEquals("inner_sym_link", status.getPath().getName());
     assertEquals(14, status.getLen());
   }
+
+  public void testUserWithMultiGroups() throws Exception {
+    Path users = new Path("/etc/users");
+    Path groups = new Path("/etc/groups_multi");
+
+    writeFile(fs, users, "tom:2001\n");
+    writeFile(fs, groups, "groupa:3001:tom\ngroupb:3002:tom");
+
+    Configuration conf = new Configuration(cfg);
+    conf.set("juicefs.users", users.toUri().getPath());
+    conf.set("juicefs.groups", groups.toUri().getPath());
+
+    FileSystem superFs = createNewFs(conf, "hdfs", new String[]{"hadoop"});
+    Path testDir = new Path("/test_multi_group/d1");
+    superFs.mkdirs(testDir);
+    superFs.setOwner(testDir.getParent(), "hdfs", "groupb");
+    superFs.setOwner(testDir, "hdfs", "groupb");
+    superFs.setPermission(testDir.getParent(), FsPermission.createImmutable((short) 0770));
+    superFs.setPermission(testDir, FsPermission.createImmutable((short) 0770));
+
+    FileSystem tomFs = createNewFs(conf, "tom", new String[]{"randgroup"});
+    tomFs.listStatus(testDir);
+
+    tomFs.close();
+    superFs.close();
+  }
 }