refactor reading and assembling permissions

Signed-off-by: Jörn Friedrich Dreyer <[email protected]>
cs3org · Dec 14, 2020 · 15f25f8 · 15f25f8
1 parent aac9af9
commit 15f25f8
Show file tree

Hide file tree

Showing 4 changed files with 100 additions and 53 deletions.
diff --git a/pkg/storage/fs/ocis/node.go b/pkg/storage/fs/ocis/node.go
@@ -306,7 +306,7 @@ func (n *Node) PermissionSet(ctx context.Context) *provider.ResourcePermissions
 	u, ok := user.ContextGetUser(ctx)
 	if !ok {
 		appctx.GetLogger(ctx).Debug().Interface("node", n).Msg("no user in context, returning default permissions")
-		return defaultPermissions
+		return noPermissions
 	}
 	n.Owner() // read owner
 	if u.Id.OpaqueId == n.ownerID && u.Id.Idp == n.ownerIDP {
@@ -315,10 +315,10 @@ func (n *Node) PermissionSet(ctx context.Context) *provider.ResourcePermissions
 	// TODO fix permissions for share recipients by traversing up to the next acl? or to the root?
 	// ouch ... if we have to read acls for every dir entry that would be n*depth additional xattr reads ...
 	// but we only need to read the parent tree once.
-	return &provider.ResourcePermissions{
-		ListContainer:   true,
-		CreateContainer: true,
+	if np, err := n.ReadUserPermissions(ctx, u); err == nil {
+		return np
 	}
+	return noPermissions
 }
 
 // AsResourceInfo return the node as CS3 ResourceInfo
@@ -476,6 +476,84 @@ func (n *Node) UnsetTempEtag() (err error) {
 	return err
 }
 
+// ReadUserPermissions will assemble the permissions for the current user on the given node without parent nodes
+func (n *Node) ReadUserPermissions(ctx context.Context, u *userpb.User) (ap *provider.ResourcePermissions, err error) {
+	// check if the current user is the owner
+	id, idp, err := n.Owner()
+	if err != nil {
+		// TODO check if a parent folder has the owner set?
+		appctx.GetLogger(ctx).Error().Err(err).Interface("node", n).Msg("could not determine owner, returning default permissions")
+		return noPermissions, err
+	}
+	if id == "" {
+		// TODO what if no owner is set but grants are present?
+		return noOwnerPermissions, nil
+	}
+	if id == u.Id.OpaqueId && idp == u.Id.Idp {
+		appctx.GetLogger(ctx).Debug().Interface("node", n).Msg("user is owner, returning owner permissions")
+		return ownerPermissions, nil
+	}
+
+	ap = &provider.ResourcePermissions{}
+
+	// for an efficient group lookup convert the list of groups to a map
+	// groups are just strings ... groupnames ... or group ids ??? AAARGH !!!
+	groupsMap := make(map[string]bool, len(u.Groups))
+	for i := range u.Groups {
+		groupsMap[u.Groups[i]] = true
+	}
+
+	var g *provider.Grant
+
+	// we read all grantees from the node
+	var grantees []string
+	if grantees, err = n.ListGrantees(ctx); err != nil {
+		appctx.GetLogger(ctx).Error().Err(err).Interface("node", n).Msg("error listing grantees")
+		return nil, err
+	}
+
+	// instead of making n getxattr syscalls we are going to list the acls and filter them here
+	// we have two options here:
+	// 1. we can start iterating over the acls / grants on the node or
+	// 2. we can iterate over the number of groups
+	// The current implementation tries to be defensive for cases where users have hundreds or thousants of groups, so we iterate over the existing acls.
+	userace := grantPrefix + _userAcePrefix + u.Id.OpaqueId
+	userFound := false
+	for i := range grantees {
+		switch {
+		// we only need to find the user once
+		case !userFound && grantees[i] == userace:
+			g, err = n.ReadGrant(ctx, grantees[i])
+		case strings.HasPrefix(grantees[i], grantPrefix+_groupAcePrefix): // only check group grantees
+			gr := strings.TrimPrefix(grantees[i], grantPrefix+_groupAcePrefix)
+			if groupsMap[gr] {
+				g, err = n.ReadGrant(ctx, grantees[i])
+			} else {
+				// no need to check attribute
+				continue
+			}
+		default:
+			// no need to check attribute
+			continue
+		}
+
+		switch {
+		case err == nil:
+			addPermissions(ap, g.GetPermissions())
+		case isNoData(err):
+			err = nil
+			appctx.GetLogger(ctx).Error().Interface("node", n).Str("grant", grantees[i]).Interface("grantees", grantees).Msg("grant vanished from node after listing")
+			// continue with next segment
+		default:
+			appctx.GetLogger(ctx).Error().Err(err).Interface("node", n).Str("grant", grantees[i]).Msg("error reading permissions")
+			// continue with next segment
+		}
+	}
+
+	appctx.GetLogger(ctx).Debug().Interface("permissions", ap).Interface("node", n).Interface("user", u).Msg("returning aggregated permissions")
+	return ap, nil
+}
+
 // ListGrantees lists the grantees of the current node
 // We don't want to wast time and memory by creating grantee objects.
 // The function will return a list of opaque strings that can be used to make a ReadGrant call

diff --git a/pkg/storage/fs/ocis/ocis.go b/pkg/storage/fs/ocis/ocis.go
@@ -360,7 +360,8 @@ func (fs *ocisfs) GetMD(ctx context.Context, ref *provider.Reference, mdKeys []s
 		err = errtypes.NotFound(filepath.Join(node.ParentID, node.Name))
 		return
 	}
-	rp, err := fs.p.ReadUserPermissions(ctx, node)
+
+	rp, err := fs.p.AssemblePermissions(ctx, node)
 	switch {
 	case err != nil:
 		return nil, errtypes.InternalError(err.Error())
@@ -382,7 +383,7 @@ func (fs *ocisfs) ListFolder(ctx context.Context, ref *provider.Reference, mdKey
 		return
 	}
 
-	rp, err := fs.p.ReadUserPermissions(ctx, node)
+	rp, err := fs.p.AssemblePermissions(ctx, node)
 	switch {
 	case err != nil:
 		return nil, errtypes.InternalError(err.Error())

diff --git a/pkg/storage/fs/ocis/permissions.go b/pkg/storage/fs/ocis/permissions.go
@@ -36,7 +36,7 @@ const (
 	_groupAcePrefix = "g:"
 )
 
-var defaultPermissions *provider.ResourcePermissions = &provider.ResourcePermissions{
+var noPermissions *provider.ResourcePermissions = &provider.ResourcePermissions{
 	// no permissions
 }
 
@@ -71,19 +71,19 @@ type Permissions struct {
 	lu *Lookup
 }
 
-// ReadUserPermissions will assemble the permissions for the current user on the given node
-func (p *Permissions) ReadUserPermissions(ctx context.Context, n *Node) (ap *provider.ResourcePermissions, err error) {
+// AssemblePermissions will assemble the permissions for the current user on the given node, taking into account all parent nodes
+func (p *Permissions) AssemblePermissions(ctx context.Context, n *Node) (ap *provider.ResourcePermissions, err error) {
 	u, ok := user.ContextGetUser(ctx)
 	if !ok {
 		appctx.GetLogger(ctx).Debug().Interface("node", n).Msg("no user in context, returning default permissions")
-		return defaultPermissions, nil
+		return noPermissions, nil
 	}
 	// check if the current user is the owner
 	id, idp, err := n.Owner()
 	if err != nil {
 		// TODO check if a parent folder has the owner set?
 		appctx.GetLogger(ctx).Error().Err(err).Interface("node", n).Msg("could not determine owner, returning default permissions")
-		return defaultPermissions, err
+		return noPermissions, err
 	}
 	if id == "" {
 		// TODO what if no owner is set but grants are present?
@@ -111,47 +111,14 @@ func (p *Permissions) ReadUserPermissions(ctx context.Context, n *Node) (ap *pro
 		groupsMap[u.Groups[i]] = true
 	}
 
-	var g *provider.Grant
 	// for all segments, starting at the leaf
 	for cn.ID != rn.ID {
 
-		var grantees []string
-		if grantees, err = cn.ListGrantees(ctx); err != nil {
-			appctx.GetLogger(ctx).Error().Err(err).Interface("node", cn).Msg("error listing grantees")
-			return nil, err
-		}
-
-		userace := grantPrefix + _userAcePrefix + u.Id.OpaqueId
-		userFound := false
-		for i := range grantees {
-			// we only need the find the user once per node
-			switch {
-			case !userFound && grantees[i] == userace:
-				g, err = cn.ReadGrant(ctx, grantees[i])
-			case strings.HasPrefix(grantees[i], grantPrefix+_groupAcePrefix):
-				gr := strings.TrimPrefix(grantees[i], grantPrefix+_groupAcePrefix)
-				if groupsMap[gr] {
-					g, err = cn.ReadGrant(ctx, grantees[i])
-				} else {
-					// no need to check attribute
-					continue
-				}
-			default:
-				// no need to check attribute
-				continue
-			}
-
-			switch {
-			case err == nil:
-				addPermissions(ap, g.GetPermissions())
-			case isNoData(err):
-				err = nil
-				appctx.GetLogger(ctx).Error().Interface("node", cn).Str("grant", grantees[i]).Interface("grantees", grantees).Msg("grant vanished from node after listing")
-				// continue with next segment
-			default:
-				appctx.GetLogger(ctx).Error().Err(err).Interface("node", cn).Str("grant", grantees[i]).Msg("error reading permissions")
-				// continue with next segment
-			}
+		if np, err := cn.ReadUserPermissions(ctx, u); err == nil {
+			addPermissions(ap, np)
+		} else {
+			appctx.GetLogger(ctx).Error().Err(err).Interface("node", cn).Msg("error reading permissions")
+			// continue with next segment
 		}
 
 		if cn, err = cn.Parent(); err != nil {
@@ -259,21 +226,21 @@ func (p *Permissions) HasPermission(ctx context.Context, n *Node, check func(*pr
 		}
 	}
 
-	appctx.GetLogger(ctx).Debug().Interface("permissions", defaultPermissions).Interface("node", n).Interface("user", u).Msg("no grant found, returning default permissions")
+	appctx.GetLogger(ctx).Debug().Interface("permissions", noPermissions).Interface("node", n).Interface("user", u).Msg("no grant found, returning default permissions")
 	return false, nil
 }
 
 func (p *Permissions) getUserAndPermissions(ctx context.Context, n *Node) (*userv1beta1.User, *provider.ResourcePermissions) {
 	u, ok := user.ContextGetUser(ctx)
 	if !ok {
 		appctx.GetLogger(ctx).Debug().Interface("node", n).Msg("no user in context, returning default permissions")
-		return nil, defaultPermissions
+		return nil, noPermissions
 	}
 	// check if the current user is the owner
 	id, _, err := n.Owner()
 	if err != nil {
 		appctx.GetLogger(ctx).Error().Err(err).Interface("node", n).Msg("could not determine owner, returning default permissions")
-		return nil, defaultPermissions
+		return nil, noPermissions
 	}
 	if id == "" {
 		// TODO what if no owner is set but grants are present?

diff --git a/pkg/storage/fs/ocis/recycle.go b/pkg/storage/fs/ocis/recycle.go
@@ -50,13 +50,14 @@ func (fs *ocisfs) ListRecycle(ctx context.Context) (items []*provider.RecycleIte
 	items = make([]*provider.RecycleItem, 0)
 
 	// TODO how do we check if the storage allows listing the recycle for the current user? check owner of the root of the storage?
+	// use permissions ReadUserPermissions?
 	if fs.o.EnableHome {
 		if !ownerPermissions.ListContainer {
 			log.Debug().Msg("owner not allowed to list trash")
 			return items, errtypes.PermissionDenied("owner not allowed to list trash")
 		}
 	} else {
-		if !defaultPermissions.ListContainer {
+		if !noPermissions.ListContainer {
 			log.Debug().Msg("default permissions prevent listing trash")
 			return items, errtypes.PermissionDenied("default permissions prevent listing trash")
 		}