Harden decomposedfs and add unit tests (cs3org#2187)

* lookup space root and tests

* refactor checkQuota

* move storageSpace lookup to node

changelog/unreleased/get-quota-storage-space.md

@@ -6,3 +6,4 @@ Make the cs3apis accept a Reference in the getQuota Request to limit the call to
 
 https://github.com/cs3org/reva/pull/2152
 https://github.com/cs3org/reva/pull/2178
+https://github.com/cs3org/reva/pull/2187

pkg/storage/utils/decomposedfs/decomposedfs.go

@@ -163,7 +163,7 @@ func (fs *Decomposedfs) GetQuota(ctx context.Context, ref *provider.Reference) (
 		quotaStr = string(ri.Opaque.Map["quota"].Value)
 	}
 
-	avail, err := fs.getAvailableSize(n.InternalPath())
+	avail, err := node.GetAvailableSize(n.InternalPath())
 	if err != nil {
 		return 0, 0, err
 	}

pkg/storage/utils/decomposedfs/lookup.go

@@ -45,11 +45,10 @@ func (lu *Lookup) NodeFromResource(ctx context.Context, ref *provider.Reference)
 	if ref.ResourceId != nil {
 		// check if a storage space reference is used
 		// currently, the decomposed fs uses the root node id as the space id
-		spaceRoot, err := lu.NodeFromID(ctx, ref.ResourceId)
+		n, err := lu.NodeFromID(ctx, ref.ResourceId)
 		if err != nil {
 			return nil, err
 		}
-		n := spaceRoot
 		// is this a relative reference?
 		if ref.Path != "" {
 			p := filepath.Clean(ref.Path)
@@ -62,8 +61,6 @@ func (lu *Lookup) NodeFromResource(ctx context.Context, ref *provider.Reference)
 					return nil, err
 				}
 			}
-			// use reference id as space root for relative references
-			n.SpaceRoot = spaceRoot
 		}
 		return n, nil
 	}
@@ -110,7 +107,12 @@ func (lu *Lookup) NodeFromID(ctx context.Context, id *provider.ResourceId) (n *n
 	if id == nil || id.OpaqueId == "" {
 		return nil, fmt.Errorf("invalid resource id %+v", id)
 	}
-	return node.ReadNode(ctx, lu, id.OpaqueId)
+	n, err = node.ReadNode(ctx, lu, id.OpaqueId)
+	if err != nil {
+		return nil, err
+	}
+
+	return n, n.FindStorageSpaceRoot()
 }
 
 // Path returns the path for node
@@ -179,6 +181,9 @@ func (lu *Lookup) WalkPath(ctx context.Context, r *node.Node, p string, followRe
 				}
 			}
 		}
+		if node.IsSpaceRoot(r) {
+			r.SpaceRoot = r
+		}
 
 		if !r.Exists && i < len(segments)-1 {
 			return r, errtypes.NotFound(segments[i])

pkg/storage/utils/decomposedfs/lookup_test.go

@@ -19,9 +19,9 @@
 package decomposedfs_test
 
 import (
+	provider "github.com/cs3org/go-cs3apis/cs3/storage/provider/v1beta1"
 	helpers "github.com/cs3org/reva/pkg/storage/utils/decomposedfs/testhelpers"
 	"github.com/cs3org/reva/pkg/storage/utils/decomposedfs/xattrs"
-
 	. "github.com/onsi/ginkgo"
 	. "github.com/onsi/gomega"
 )
@@ -41,10 +41,9 @@ var _ = Describe("Lookup", func() {
 		if env != nil {
 			env.Cleanup()
 		}
-
 	})
 
-	Describe("Path", func() {
+	Describe("Node from path", func() {
 		It("returns the path including a leading slash", func() {
 			n, err := env.Lookup.NodeFromPath(env.Ctx, "/dir1/file1", false)
 			Expect(err).ToNot(HaveOccurred())
@@ -55,6 +54,57 @@ var _ = Describe("Lookup", func() {
 		})
 	})
 
+	Describe("Node From Resource only by path", func() {
+		It("returns the path including a leading slash and the space root is set", func() {
+			n, err := env.Lookup.NodeFromResource(env.Ctx, &provider.Reference{Path: "/dir1/subdir1/file2"})
+			Expect(err).ToNot(HaveOccurred())
+
+			path, err := env.Lookup.Path(env.Ctx, n)
+			Expect(err).ToNot(HaveOccurred())
+			Expect(path).To(Equal("/dir1/subdir1/file2"))
+			Expect(n.SpaceRoot.Name).To(Equal("userid"))
+			Expect(n.SpaceRoot.ParentID).To(Equal("root"))
+		})
+	})
+
+	Describe("Node From Resource only by id", func() {
+		It("returns the path including a leading slash and the space root is set", func() {
+			// do a node lookup by path
+			nRef, err := env.Lookup.NodeFromPath(env.Ctx, "/dir1/file1", false)
+			Expect(err).ToNot(HaveOccurred())
+
+			// try to find the same node by id
+			n, err := env.Lookup.NodeFromResource(env.Ctx, &provider.Reference{ResourceId: &provider.ResourceId{OpaqueId: nRef.ID}})
+			Expect(err).ToNot(HaveOccurred())
+
+			// Check if we got the right node and spaceRoot
+			path, err := env.Lookup.Path(env.Ctx, n)
+			Expect(err).ToNot(HaveOccurred())
+			Expect(path).To(Equal("/dir1/file1"))
+			Expect(n.SpaceRoot.Name).To(Equal("userid"))
+			Expect(n.SpaceRoot.ParentID).To(Equal("root"))
+		})
+	})
+
+	Describe("Node From Resource by id and relative path", func() {
+		It("returns the path including a leading slash and the space root is set", func() {
+			// do a node lookup by path for the parent
+			nRef, err := env.Lookup.NodeFromPath(env.Ctx, "/dir1", false)
+			Expect(err).ToNot(HaveOccurred())
+
+			// try to find the child node by parent id and relative path
+			n, err := env.Lookup.NodeFromResource(env.Ctx, &provider.Reference{ResourceId: &provider.ResourceId{OpaqueId: nRef.ID}, Path: "./file1"})
+			Expect(err).ToNot(HaveOccurred())
+
+			// Check if we got the right node and spaceRoot
+			path, err := env.Lookup.Path(env.Ctx, n)
+			Expect(err).ToNot(HaveOccurred())
+			Expect(path).To(Equal("/dir1/file1"))
+			Expect(n.SpaceRoot.Name).To(Equal("userid"))
+			Expect(n.SpaceRoot.ParentID).To(Equal("root"))
+		})
+	})
+
 	Describe("Reference Parsing", func() {
 		It("parses a valid cs3 reference", func() {
 			in := []byte("cs3:bede11a0-ea3d-11eb-a78b-bf907adce8ed/c402d01c-ea3d-11eb-a0fc-c32f9d32528f")

pkg/storage/utils/decomposedfs/node/node.go

@@ -192,6 +192,7 @@ func ReadNode(ctx context.Context, lu PathLookup, id string) (n *Node, err error
 	default:
 		return nil, errtypes.InternalError(err.Error())
 	}
+
 	// check if this is a space root
 	if _, err = xattr.Get(nodePath, xattrs.SpaceNameAttr); err == nil {
 		n.SpaceRoot = n
@@ -261,6 +262,7 @@ func (n *Node) Child(ctx context.Context, name string) (*Node, error) {
 		if err != nil {
 			return nil, errors.Wrap(err, "could not read child node")
 		}
+		c.SpaceRoot = n.SpaceRoot
 	} else {
 		return nil, fmt.Errorf("Decomposedfs: expected '../ prefix, got' %+v", link)
 	}
@@ -936,3 +938,60 @@ func parseMTime(v string) (t time.Time, err error) {
 	}
 	return time.Unix(sec, nsec), err
 }
+
+// FindStorageSpaceRoot calls n.Parent() and climbs the tree
+// until it finds the space root node and adds it to the node
+func (n *Node) FindStorageSpaceRoot() error {
+	var err error
+	// remember the node we ask for and use parent to climb the tree
+	parent := n
+	for parent.ParentID != "" {
+		if parent, err = parent.Parent(); err != nil {
+			return err
+		}
+		if IsSpaceRoot(parent) {
+			n.SpaceRoot = parent
+			break
+		}
+	}
+	return nil
+}
+
+// IsSpaceRoot checks if the node is a space root
+func IsSpaceRoot(r *Node) bool {
+	path := r.InternalPath()
+	if spaceNameBytes, err := xattr.Get(path, xattrs.SpaceNameAttr); err == nil {
+		if string(spaceNameBytes) != "" {
+			return true
+		}
+	}
+	return false
+}
+
+// CheckQuota checks if both disk space and available quota are sufficient
+var CheckQuota = func(spaceRoot *Node, fileSize uint64) (quotaSufficient bool, err error) {
+	used, _ := spaceRoot.GetTreeSize()
+	if !enoughDiskSpace(spaceRoot.InternalPath(), fileSize) {
+		return false, errtypes.InsufficientStorage("disk full")
+	}
+	quotaByte, _ := xattr.Get(spaceRoot.InternalPath(), xattrs.QuotaAttr)
+	var total uint64
+	if quotaByte == nil {
+		// if quota is not set, it means unlimited
+		return true, nil
+	}
+	total, _ = strconv.ParseUint(string(quotaByte), 10, 64)
+	// if total is smaller than used, total-used could overflow and be bigger than fileSize
+	if fileSize > total-used || total < used {
+		return false, errtypes.InsufficientStorage("quota exceeded")
+	}
+	return true, nil
+}
+
+func enoughDiskSpace(path string, fileSize uint64) bool {
+	avalB, err := GetAvailableSize(path)
+	if err != nil {
+		return false
+	}
+	return avalB > fileSize
+}

pkg/storage/utils/decomposedfs/decomposedfs_unix.go → pkg/storage/utils/decomposedfs/node/node_unix.go

@@ -19,11 +19,12 @@
 //go:build !windows
 // +build !windows
 
-package decomposedfs
+package node
 
 import "syscall"
 
-func (fs *Decomposedfs) getAvailableSize(path string) (uint64, error) {
+// GetAvailableSize stats the filesystem and return the available bytes
+func GetAvailableSize(path string) (uint64, error) {
 	stat := syscall.Statfs_t{}
 	err := syscall.Statfs(path, &stat)
 	if err != nil {

pkg/storage/utils/decomposedfs/decomposedfs_windows.go → pkg/storage/utils/decomposedfs/node/node_windows.go

@@ -19,11 +19,12 @@
 //go:build windows
 // +build windows
 
-package decomposedfs
+package node
 
 import "golang.org/x/sys/windows"
 
-func (fs *Decomposedfs) getAvailableSize(path string) (uint64, error) {
+// GetAvailableSize stats the filesystem and return the available bytes
+func GetAvailableSize(path string) (uint64, error) {
 	var free, total, avail uint64
 	pathPtr, err := windows.UTF16PtrFromString(path)
 	if err != nil {

pkg/storage/utils/decomposedfs/testhelpers/helpers.go

@@ -23,7 +23,9 @@ import (
 	"os"
 	"path/filepath"
 
+	"github.com/cs3org/reva/pkg/storage/utils/decomposedfs/xattrs"
 	"github.com/google/uuid"
+	"github.com/pkg/xattr"
 	"github.com/stretchr/testify/mock"
 
 	userpb "github.com/cs3org/go-cs3apis/cs3/identity/user/v1beta1"
@@ -112,6 +114,15 @@ func NewTestEnv() (*TestEnv, error) {
 		return nil, err
 	}
 
+	// the space name attribute is the stop condition in the lookup
+	h, err := lookup.HomeNode(ctx)
+	if err != nil {
+		return nil, err
+	}
+	if err = xattr.Set(h.InternalPath(), xattrs.SpaceNameAttr, []byte("username")); err != nil {
+		return nil, err
+	}
+
 	// Create dir1
 	dir1, err := env.CreateTestDir("/dir1")
 	if err != nil {
@@ -130,6 +141,16 @@ func NewTestEnv() (*TestEnv, error) {
 		return nil, err
 	}
 
+	dir2, err := dir1.Child(ctx, "subdir1")
+	if err != nil {
+		return nil, err
+	}
+	// Create file1 in dir1
+	_, err = env.CreateTestFile("file2", "file2-blobid", 12345, dir2.ID)
+	if err != nil {
+		return nil, err
+	}
+
 	// Create emptydir
 	err = fs.CreateDir(ctx, &providerv1beta1.Reference{Path: "/emptydir"})
 	if err != nil {

pkg/storage/utils/decomposedfs/upload.go

@@ -31,7 +31,6 @@ import (
 	"io/ioutil"
 	"os"
 	"path/filepath"
-	"strconv"
 	"strings"
 	"time"
 
@@ -43,11 +42,9 @@ import (
 	"github.com/cs3org/reva/pkg/logger"
 	"github.com/cs3org/reva/pkg/storage/utils/chunking"
 	"github.com/cs3org/reva/pkg/storage/utils/decomposedfs/node"
-	"github.com/cs3org/reva/pkg/storage/utils/decomposedfs/xattrs"
 	"github.com/cs3org/reva/pkg/utils"
 	"github.com/google/uuid"
 	"github.com/pkg/errors"
-	"github.com/pkg/xattr"
 	"github.com/rs/zerolog"
 	tusd "github.com/tus/tusd/pkg/handler"
 )
@@ -163,7 +160,7 @@ func (fs *Decomposedfs) InitiateUpload(ctx context.Context, ref *provider.Refere
 
 	log.Debug().Interface("info", info).Interface("node", n).Interface("metadata", metadata).Msg("Decomposedfs: resolved filename")
 
-	_, err = checkQuota(ctx, fs, n.SpaceRoot, uint64(info.Size))
+	_, err = node.CheckQuota(n.SpaceRoot, uint64(info.Size))
 	if err != nil {
 		return nil, err
 	}
@@ -486,7 +483,7 @@ func (upload *fileUpload) FinishUpload(ctx context.Context) (err error) {
 	)
 	n.SpaceRoot = node.New(upload.info.Storage["SpaceRoot"], "", "", 0, "", nil, upload.fs.lu)
 
-	_, err = checkQuota(upload.ctx, upload.fs, n.SpaceRoot, uint64(fi.Size()))
+	_, err = node.CheckQuota(n.SpaceRoot, uint64(fi.Size()))
 	if err != nil {
 		return err
 	}
@@ -749,33 +746,3 @@ func (upload *fileUpload) ConcatUploads(ctx context.Context, uploads []tusd.Uplo
 
 	return
 }
-
-func checkQuota(ctx context.Context, fs *Decomposedfs, spaceRoot *node.Node, fileSize uint64) (quotaSufficient bool, err error) {
-	used, _ := spaceRoot.GetTreeSize()
-	enoughDiskSpace := enoughDiskSpace(fs, spaceRoot.InternalPath(), fileSize)
-	if !enoughDiskSpace {
-		return false, errtypes.InsufficientStorage("disk full")
-	}
-	quotaB, _ := xattr.Get(spaceRoot.InternalPath(), xattrs.QuotaAttr)
-	var total uint64
-	if quotaB != nil {
-		total, _ = strconv.ParseUint(string(quotaB), 10, 64)
-	} else {
-		// if quota is not set, it means unlimited
-		return true, nil
-	}
-
-	if fileSize > total-used || total < used {
-		return false, errtypes.InsufficientStorage("quota exceeded")
-	}
-	return true, nil
-}
-
-func enoughDiskSpace(fs *Decomposedfs, path string, fileSize uint64) bool {
-	avalB, err := fs.getAvailableSize(path)
-	if err != nil {
-		return false
-	}
-
-	return avalB > fileSize
-}