Use the opencloud reva from now on

This commit is contained in:
André Duffeck
2025-01-21 11:16:38 +01:00
parent 2c1afafb35
commit e8d35e1280
1007 changed files with 2988 additions and 27822 deletions
+28
View File
@@ -0,0 +1,28 @@
// Copyright 2018-2021 CERN
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
//
// In applying this license, CERN does not waive the privileges and immunities
// granted to it by virtue of its status as an Intergovernmental Organization
// or submit itself to any jurisdiction.
package cache
import (
provider "github.com/cs3org/go-cs3apis/cs3/storage/provider/v1beta1"
)
// Warmup is the interface to implement cache warmup strategies.
type Warmup interface {
GetResourceInfos() ([]*provider.ResourceInfo, error)
}
@@ -0,0 +1,155 @@
// Copyright 2018-2021 CERN
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
//
// In applying this license, CERN does not waive the privileges and immunities
// granted to it by virtue of its status as an Intergovernmental Organization
// or submit itself to any jurisdiction.
package cbox
import (
"context"
"database/sql"
"fmt"
userpb "github.com/cs3org/go-cs3apis/cs3/identity/user/v1beta1"
rpc "github.com/cs3org/go-cs3apis/cs3/rpc/v1beta1"
provider "github.com/cs3org/go-cs3apis/cs3/storage/provider/v1beta1"
"github.com/mitchellh/mapstructure"
"github.com/opencloud-eu/reva/v2/pkg/auth/scope"
ctxpkg "github.com/opencloud-eu/reva/v2/pkg/ctx"
"github.com/opencloud-eu/reva/v2/pkg/rgrpc/todo/pool"
"github.com/opencloud-eu/reva/v2/pkg/share/cache"
"github.com/opencloud-eu/reva/v2/pkg/share/cache/warmup/registry"
"github.com/opencloud-eu/reva/v2/pkg/token/manager/jwt"
"github.com/pkg/errors"
"google.golang.org/grpc/metadata"
// Provides mysql drivers
_ "github.com/go-sql-driver/mysql"
)
func init() {
registry.Register("cbox", New)
}
type config struct {
DbUsername string `mapstructure:"db_username"`
DbPassword string `mapstructure:"db_password"`
DbHost string `mapstructure:"db_host"`
DbPort int `mapstructure:"db_port"`
DbName string `mapstructure:"db_name"`
EOSNamespace string `mapstructure:"namespace"`
GatewaySvc string `mapstructure:"gatewaysvc"`
JWTSecret string `mapstructure:"jwt_secret"`
}
type manager struct {
conf *config
db *sql.DB
}
func parseConfig(m map[string]interface{}) (*config, error) {
c := &config{}
if err := mapstructure.Decode(m, c); err != nil {
err = errors.Wrap(err, "error decoding conf")
return nil, err
}
return c, nil
}
// New returns an implementation of cache warmup that connects to the cbox share db and stats resources on EOS
func New(m map[string]interface{}) (cache.Warmup, error) {
c, err := parseConfig(m)
if err != nil {
return nil, err
}
db, err := sql.Open("mysql", fmt.Sprintf("%s:%s@tcp(%s:%d)/%s", c.DbUsername, c.DbPassword, c.DbHost, c.DbPort, c.DbName))
if err != nil {
return nil, err
}
return &manager{
conf: c,
db: db,
}, nil
}
func (m *manager) GetResourceInfos() ([]*provider.ResourceInfo, error) {
query := "select coalesce(fileid_prefix, '') as fileid_prefix, coalesce(item_source, '') as item_source FROM oc_share WHERE (orphan = 0 or orphan IS NULL)"
rows, err := m.db.Query(query)
if err != nil {
return nil, err
}
defer rows.Close()
tokenManager, err := jwt.New(map[string]interface{}{
"secret": m.conf.JWTSecret,
})
if err != nil {
return nil, err
}
u := &userpb.User{
Id: &userpb.UserId{
OpaqueId: "root",
},
UidNumber: 0,
GidNumber: 0,
}
scope, err := scope.AddOwnerScope(nil)
if err != nil {
return nil, err
}
tkn, err := tokenManager.MintToken(context.Background(), u, scope)
if err != nil {
return nil, err
}
ctx := metadata.AppendToOutgoingContext(context.Background(), ctxpkg.TokenHeader, tkn)
client, err := pool.GetGatewayServiceClient(m.conf.GatewaySvc)
if err != nil {
return nil, err
}
infos := []*provider.ResourceInfo{}
for rows.Next() {
var spaceID, nodeID string
if err := rows.Scan(&spaceID, &nodeID); err != nil {
continue
}
statReq := provider.StatRequest{Ref: &provider.Reference{
ResourceId: &provider.ResourceId{
SpaceId: spaceID,
OpaqueId: nodeID,
},
}}
statRes, err := client.Stat(ctx, &statReq)
if err != nil || statRes.Status.Code != rpc.Code_CODE_OK {
continue
}
infos = append(infos, statRes.Info)
}
if err = rows.Err(); err != nil {
return nil, err
}
return infos, nil
}
@@ -0,0 +1,25 @@
// Copyright 2018-2021 CERN
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
//
// In applying this license, CERN does not waive the privileges and immunities
// granted to it by virtue of its status as an Intergovernmental Organization
// or submit itself to any jurisdiction.
package loader
import (
// Load share cache drivers.
_ "github.com/opencloud-eu/reva/v2/pkg/share/cache/warmup/cbox"
// Add your own here
)
@@ -0,0 +1,34 @@
// Copyright 2018-2021 CERN
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
//
// In applying this license, CERN does not waive the privileges and immunities
// granted to it by virtue of its status as an Intergovernmental Organization
// or submit itself to any jurisdiction.
package registry
import "github.com/opencloud-eu/reva/v2/pkg/share/cache"
// NewFunc is the function that cache warmup implementations
// should register at init time.
type NewFunc func(map[string]interface{}) (cache.Warmup, error)
// NewFuncs is a map containing all the registered cache warmup implementations.
var NewFuncs = map[string]NewFunc{}
// Register registers a new cache warmup function.
// Not safe for concurrent use. Safe for use from package init.
func Register(name string, f NewFunc) {
NewFuncs[name] = f
}
+887
View File
@@ -0,0 +1,887 @@
// Copyright 2018-2022 CERN
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
//
// In applying this license, CERN does not waive the privileges and immunities
// granted to it by virtue of its status as an Intergovernmental Organization
// or submit itself to any jurisdiction.
package cs3
import (
"context"
"encoding/json"
"fmt"
"net/url"
"path"
"strings"
"sync"
gatewayv1beta1 "github.com/cs3org/go-cs3apis/cs3/gateway/v1beta1"
groupv1beta1 "github.com/cs3org/go-cs3apis/cs3/identity/group/v1beta1"
userpb "github.com/cs3org/go-cs3apis/cs3/identity/user/v1beta1"
rpcv1beta1 "github.com/cs3org/go-cs3apis/cs3/rpc/v1beta1"
collaboration "github.com/cs3org/go-cs3apis/cs3/sharing/collaboration/v1beta1"
provider "github.com/cs3org/go-cs3apis/cs3/storage/provider/v1beta1"
"github.com/google/uuid"
"github.com/mitchellh/mapstructure"
"github.com/opencloud-eu/reva/v2/pkg/appctx"
ctxpkg "github.com/opencloud-eu/reva/v2/pkg/ctx"
"github.com/opencloud-eu/reva/v2/pkg/errtypes"
"github.com/opencloud-eu/reva/v2/pkg/rgrpc/todo/pool"
"github.com/opencloud-eu/reva/v2/pkg/share"
"github.com/opencloud-eu/reva/v2/pkg/share/manager/registry"
"github.com/opencloud-eu/reva/v2/pkg/storage/utils/indexer"
indexerErrors "github.com/opencloud-eu/reva/v2/pkg/storage/utils/indexer/errors"
"github.com/opencloud-eu/reva/v2/pkg/storage/utils/indexer/option"
"github.com/opencloud-eu/reva/v2/pkg/storage/utils/metadata"
"github.com/opencloud-eu/reva/v2/pkg/storagespace"
"github.com/opencloud-eu/reva/v2/pkg/utils"
"github.com/pkg/errors"
"google.golang.org/genproto/protobuf/field_mask"
)
// Manager implements a share manager using a cs3 storage backend
type Manager struct {
gatewayClient gatewayv1beta1.GatewayAPIClient
sync.RWMutex
storage metadata.Storage
indexer indexer.Indexer
initialized bool
}
// ReceivedShareMetadata hold the state information or a received share
type ReceivedShareMetadata struct {
State collaboration.ShareState `json:"state"`
MountPoint *provider.Reference `json:"mountpoint"`
}
func init() {
registry.Register("cs3", NewDefault)
}
type config struct {
GatewayAddr string `mapstructure:"gateway_addr"`
ProviderAddr string `mapstructure:"provider_addr"`
ServiceUserID string `mapstructure:"service_user_id"`
ServiceUserIdp string `mapstructure:"service_user_idp"`
MachineAuthAPIKey string `mapstructure:"machine_auth_apikey"`
}
// NewDefault returns a new manager instance with default dependencies
func NewDefault(m map[string]interface{}) (share.Manager, error) {
c := &config{}
if err := mapstructure.Decode(m, c); err != nil {
err = errors.Wrap(err, "error creating a new manager")
return nil, err
}
s, err := metadata.NewCS3Storage(c.GatewayAddr, c.ProviderAddr, c.ServiceUserID, c.ServiceUserIdp, c.MachineAuthAPIKey)
if err != nil {
return nil, err
}
indexer := indexer.CreateIndexer(s)
client, err := pool.GetGatewayServiceClient(c.GatewayAddr)
if err != nil {
return nil, err
}
return New(client, s, indexer)
}
// New returns a new manager instance
func New(gatewayClient gatewayv1beta1.GatewayAPIClient, s metadata.Storage, indexer indexer.Indexer) (*Manager, error) {
return &Manager{
gatewayClient: gatewayClient,
storage: s,
indexer: indexer,
initialized: false,
}, nil
}
func (m *Manager) initialize() error {
if m.initialized {
return nil
}
m.Lock()
defer m.Unlock()
if m.initialized { // check if initialization happened while grabbing the lock
return nil
}
err := m.storage.Init(context.Background(), "cs3-share-manager-metadata")
if err != nil {
return err
}
if err := m.storage.MakeDirIfNotExist(context.Background(), "shares"); err != nil {
return err
}
if err := m.storage.MakeDirIfNotExist(context.Background(), "metadata"); err != nil {
return err
}
err = m.indexer.AddIndex(&collaboration.Share{}, option.IndexByFunc{
Name: "OwnerId",
Func: indexOwnerFunc,
}, "Id.OpaqueId", "shares", "non_unique", nil, true)
if err != nil {
return err
}
err = m.indexer.AddIndex(&collaboration.Share{}, option.IndexByFunc{
Name: "CreatorId",
Func: indexCreatorFunc,
}, "Id.OpaqueId", "shares", "non_unique", nil, true)
if err != nil {
return err
}
err = m.indexer.AddIndex(&collaboration.Share{}, option.IndexByFunc{
Name: "GranteeId",
Func: indexGranteeFunc,
}, "Id.OpaqueId", "shares", "non_unique", nil, true)
if err != nil {
return err
}
err = m.indexer.AddIndex(&collaboration.Share{}, option.IndexByFunc{
Name: "ResourceId",
Func: indexResourceIDFunc,
}, "Id.OpaqueId", "shares", "non_unique", nil, true)
if err != nil {
return err
}
m.initialized = true
return nil
}
// Load imports shares and received shares from channels (e.g. during migration)
func (m *Manager) Load(ctx context.Context, shareChan <-chan *collaboration.Share, receivedShareChan <-chan share.ReceivedShareWithUser) error {
log := appctx.GetLogger(ctx)
if err := m.initialize(); err != nil {
return err
}
var mu sync.Mutex
var wg sync.WaitGroup
wg.Add(2)
go func() {
for s := range shareChan {
if s == nil {
continue
}
mu.Lock()
if err := m.persistShare(context.Background(), s); err != nil {
log.Error().Err(err).Interface("share", s).Msg("error persisting share")
}
mu.Unlock()
}
wg.Done()
}()
go func() {
for s := range receivedShareChan {
if s.ReceivedShare != nil && s.UserID != nil {
mu.Lock()
if err := m.persistReceivedShare(context.Background(), s.UserID, s.ReceivedShare); err != nil {
log.Error().Err(err).Interface("received share", s).Msg("error persisting received share")
}
mu.Unlock()
}
}
wg.Done()
}()
wg.Wait()
return nil
}
func (m *Manager) getMetadata(ctx context.Context, shareid, grantee string) ReceivedShareMetadata {
// use default values if the grantee didn't configure anything yet
metadata := ReceivedShareMetadata{
State: collaboration.ShareState_SHARE_STATE_PENDING,
}
data, err := m.storage.SimpleDownload(ctx, path.Join("metadata", shareid, grantee))
if err != nil {
return metadata
}
err = json.Unmarshal(data, &metadata)
if err != nil {
appctx.GetLogger(ctx).Error().Err(err).Str("shareid", shareid).Msg("error fetching share")
}
return metadata
}
// Dump exports shares and received shares to channels (e.g. during migration)
func (m *Manager) Dump(ctx context.Context, shareChan chan<- *collaboration.Share, receivedShareChan chan<- share.ReceivedShareWithUser) error {
log := appctx.GetLogger(ctx)
if err := m.initialize(); err != nil {
return err
}
shareids, err := m.storage.ReadDir(ctx, "shares")
if err != nil {
return err
}
for _, shareid := range shareids {
var s *collaboration.Share
if s, err = m.getShareByID(ctx, shareid); err != nil {
log.Error().Err(err).Str("shareid", shareid).Msg("error fetching share")
continue
}
// dump share data
shareChan <- s
// dump grantee metadata that includes share state and mount path
grantees, err := m.storage.ReadDir(ctx, path.Join("metadata", s.Id.OpaqueId))
if err != nil {
continue
}
for _, grantee := range grantees {
metadata := m.getMetadata(ctx, s.GetId().GetOpaqueId(), grantee)
g, err := indexToGrantee(grantee)
if err != nil || g.Type != provider.GranteeType_GRANTEE_TYPE_USER {
// ignore group grants, as every user has his own received state
continue
}
receivedShareChan <- share.ReceivedShareWithUser{
UserID: g.GetUserId(),
ReceivedShare: &collaboration.ReceivedShare{
Share: s,
State: metadata.State,
MountPoint: metadata.MountPoint,
},
}
}
}
return nil
}
// Share creates a new share
func (m *Manager) Share(ctx context.Context, md *provider.ResourceInfo, g *collaboration.ShareGrant) (*collaboration.Share, error) {
if err := m.initialize(); err != nil {
return nil, err
}
user := ctxpkg.ContextMustGetUser(ctx)
// do not allow share to myself or the owner if share is for a user
if g.Grantee.Type == provider.GranteeType_GRANTEE_TYPE_USER &&
(utils.UserEqual(g.Grantee.GetUserId(), user.Id) || utils.UserEqual(g.Grantee.GetUserId(), md.Owner)) {
return nil, errtypes.BadRequest("cs3: owner/creator and grantee are the same")
}
ts := utils.TSNow()
share := &collaboration.Share{
Id: &collaboration.ShareId{
OpaqueId: uuid.NewString(),
},
ResourceId: md.Id,
Permissions: g.Permissions,
Grantee: g.Grantee,
Owner: md.Owner,
Creator: user.Id,
Ctime: ts,
Mtime: ts,
}
err := m.persistShare(ctx, share)
return share, err
}
func (m *Manager) persistShare(ctx context.Context, share *collaboration.Share) error {
data, err := json.Marshal(share)
if err != nil {
return err
}
err = m.storage.SimpleUpload(ctx, shareFilename(share.Id.OpaqueId), data)
if err != nil {
return err
}
metadataPath := path.Join("metadata", share.Id.OpaqueId)
err = m.storage.MakeDirIfNotExist(ctx, metadataPath)
if err != nil {
return err
}
_, err = m.indexer.Add(share)
if _, ok := err.(*indexerErrors.AlreadyExistsErr); ok {
return nil
}
return err
}
// GetShare gets the information for a share by the given ref.
func (m *Manager) GetShare(ctx context.Context, ref *collaboration.ShareReference) (*collaboration.Share, error) {
err := m.initialize()
if err != nil {
return nil, err
}
var s *collaboration.Share
switch {
case ref.GetId() != nil:
s, err = m.getShareByID(ctx, ref.GetId().OpaqueId)
case ref.GetKey() != nil:
s, err = m.getShareByKey(ctx, ref.GetKey())
default:
return nil, errtypes.BadRequest("neither share id nor key was given")
}
if err != nil {
return nil, err
}
// check if we are the owner or the grantee
user := ctxpkg.ContextMustGetUser(ctx)
if user.GetId().GetType() == userpb.UserType_USER_TYPE_SERVICE || share.IsCreatedByUser(s, user) || share.IsGrantedToUser(s, user) {
return s, nil
}
return nil, errtypes.NotFound("not found")
}
// Unshare deletes the share pointed by ref.
func (m *Manager) Unshare(ctx context.Context, ref *collaboration.ShareReference) error {
if err := m.initialize(); err != nil {
return err
}
share, err := m.GetShare(ctx, ref)
if err != nil {
return err
}
err = m.storage.Delete(ctx, shareFilename(ref.GetId().OpaqueId))
if err != nil {
if _, ok := err.(errtypes.NotFound); !ok {
return err
}
}
return m.indexer.Delete(share)
}
// ListShares returns the shares created by the user
func (m *Manager) ListShares(ctx context.Context, filters []*collaboration.Filter) ([]*collaboration.Share, error) {
if err := m.initialize(); err != nil {
return nil, err
}
user, ok := ctxpkg.ContextGetUser(ctx)
if !ok {
return nil, errtypes.UserRequired("error getting user from context")
}
var rIDs []*provider.ResourceId
if len(filters) != 0 {
grouped := share.GroupFiltersByType(filters)
for _, g := range grouped {
for _, f := range g {
if f.GetResourceId() != nil {
rIDs = append(rIDs, f.GetResourceId())
}
}
}
}
var (
createdShareIds []string
err error
)
// in spaces, always use the resourceId
// We could have more than one resourceID
// which would form a logical OR
if len(rIDs) != 0 {
for _, rID := range rIDs {
shareIDs, err := m.indexer.FindBy(&collaboration.Share{},
indexer.NewField("ResourceId", resourceIDToIndex(rID)),
)
if err != nil {
return nil, err
}
createdShareIds = append(createdShareIds, shareIDs...)
}
} else {
createdShareIds, err = m.indexer.FindBy(&collaboration.Share{},
indexer.NewField("OwnerId", userIDToIndex(user.Id)),
indexer.NewField("CreatorId", userIDToIndex(user.Id)),
)
if err != nil {
return nil, err
}
}
// We use shareMem as a temporary lookup store to check which shares were
// already added. This is to prevent duplicates.
shareMem := make(map[string]struct{})
result := []*collaboration.Share{}
for _, id := range createdShareIds {
s, err := m.getShareByID(ctx, id)
if err != nil {
return nil, err
}
if share.MatchesFilters(s, filters) {
result = append(result, s)
shareMem[s.Id.OpaqueId] = struct{}{}
}
}
// If a user requests to list shares which have not been created by them
// we have to explicitly fetch these shares and check if the user is
// allowed to list the shares.
// Only then can we add these shares to the result.
grouped := share.GroupFiltersByType(filters)
idFilter, ok := grouped[collaboration.Filter_TYPE_RESOURCE_ID]
if !ok {
return result, nil
}
shareIDsByResourceID := make(map[string]*provider.ResourceId)
for _, filter := range idFilter {
resourceID := filter.GetResourceId()
shareIDs, err := m.indexer.FindBy(&collaboration.Share{},
indexer.NewField("ResourceId", resourceIDToIndex(resourceID)),
)
if err != nil {
continue
}
for _, shareID := range shareIDs {
shareIDsByResourceID[shareID] = resourceID
}
}
// statMem is used as a local cache to prevent statting resources which
// already have been checked.
statMem := make(map[string]struct{})
for shareID, resourceID := range shareIDsByResourceID {
if _, handled := shareMem[shareID]; handled {
// We don't want to add a share multiple times when we added it
// already.
continue
}
if _, checked := statMem[resourceIDToIndex(resourceID)]; !checked {
sReq := &provider.StatRequest{
Ref: &provider.Reference{ResourceId: resourceID},
}
sRes, err := m.gatewayClient.Stat(ctx, sReq)
if err != nil {
continue
}
if sRes.Status.Code != rpcv1beta1.Code_CODE_OK {
continue
}
if !sRes.Info.PermissionSet.ListGrants {
continue
}
statMem[resourceIDToIndex(resourceID)] = struct{}{}
}
s, err := m.getShareByID(ctx, shareID)
if err != nil {
return nil, err
}
if share.MatchesFilters(s, filters) {
result = append(result, s)
shareMem[s.Id.OpaqueId] = struct{}{}
}
}
return result, nil
}
// UpdateShare updates the mode of the given share.
func (m *Manager) UpdateShare(ctx context.Context, ref *collaboration.ShareReference, p *collaboration.SharePermissions, updated *collaboration.Share, fieldMask *field_mask.FieldMask) (*collaboration.Share, error) {
if err := m.initialize(); err != nil {
return nil, err
}
share, err := m.GetShare(ctx, ref)
if err != nil {
return nil, err
}
share.Permissions = p
data, err := json.Marshal(share)
if err != nil {
return nil, err
}
err = m.storage.SimpleUpload(ctx, shareFilename(share.Id.OpaqueId), data)
return share, err
}
// ListReceivedShares returns the list of shares the user has access to.
func (m *Manager) ListReceivedShares(ctx context.Context, filters []*collaboration.Filter, forUser *userpb.UserId) ([]*collaboration.ReceivedShare, error) {
if err := m.initialize(); err != nil {
return nil, err
}
user, ok := ctxpkg.ContextGetUser(ctx)
if !ok {
return nil, errtypes.UserRequired("error getting user from context")
}
uid, groups := user.GetId(), user.GetGroups()
if user.GetId().GetType() == userpb.UserType_USER_TYPE_SERVICE {
u, err := utils.GetUser(forUser, m.gatewayClient)
if err != nil {
return nil, errtypes.BadRequest("user not found")
}
uid = forUser
groups = u.GetGroups()
}
result := []*collaboration.ReceivedShare{}
ids, err := granteeToIndex(&provider.Grantee{
Type: provider.GranteeType_GRANTEE_TYPE_USER,
Id: &provider.Grantee_UserId{UserId: uid},
})
if err != nil {
return nil, err
}
receivedIds, err := m.indexer.FindBy(&collaboration.Share{},
indexer.NewField("GranteeId", ids),
)
if err != nil {
return nil, err
}
for _, group := range groups {
index, err := granteeToIndex(&provider.Grantee{
Type: provider.GranteeType_GRANTEE_TYPE_GROUP,
Id: &provider.Grantee_GroupId{GroupId: &groupv1beta1.GroupId{OpaqueId: group}},
})
if err != nil {
return nil, err
}
groupIds, err := m.indexer.FindBy(&collaboration.Share{},
indexer.NewField("GranteeId", index),
)
if err != nil {
return nil, err
}
receivedIds = append(receivedIds, groupIds...)
}
for _, id := range receivedIds {
s, err := m.getShareByID(ctx, id)
if err != nil {
return nil, err
}
if !share.MatchesFilters(s, filters) {
continue
}
metadata, err := m.downloadMetadata(ctx, s)
if err != nil {
if _, ok := err.(errtypes.NotFound); !ok {
return nil, err
}
// use default values if the grantee didn't configure anything yet
metadata = ReceivedShareMetadata{
State: collaboration.ShareState_SHARE_STATE_PENDING,
}
}
result = append(result, &collaboration.ReceivedShare{
Share: s,
State: metadata.State,
MountPoint: metadata.MountPoint,
})
}
return result, nil
}
// GetReceivedShare returns the information for a received share.
func (m *Manager) GetReceivedShare(ctx context.Context, ref *collaboration.ShareReference) (*collaboration.ReceivedShare, error) {
if err := m.initialize(); err != nil {
return nil, err
}
share, err := m.GetShare(ctx, ref)
if err != nil {
return nil, err
}
metadata, err := m.downloadMetadata(ctx, share)
if err != nil {
if _, ok := err.(errtypes.NotFound); !ok {
return nil, err
}
// use default values if the grantee didn't configure anything yet
metadata = ReceivedShareMetadata{
State: collaboration.ShareState_SHARE_STATE_PENDING,
}
}
return &collaboration.ReceivedShare{
Share: share,
State: metadata.State,
MountPoint: metadata.MountPoint,
}, nil
}
// UpdateReceivedShare updates the received share with share state.
func (m *Manager) UpdateReceivedShare(ctx context.Context, rshare *collaboration.ReceivedShare, fieldMask *field_mask.FieldMask, forUser *userpb.UserId) (*collaboration.ReceivedShare, error) {
if err := m.initialize(); err != nil {
return nil, err
}
user, ok := ctxpkg.ContextGetUser(ctx)
if !ok {
return nil, errtypes.UserRequired("error getting user from context")
}
rs, err := m.GetReceivedShare(ctx, &collaboration.ShareReference{Spec: &collaboration.ShareReference_Id{Id: rshare.Share.Id}})
if err != nil {
return nil, err
}
for i := range fieldMask.Paths {
switch fieldMask.Paths[i] {
case "state":
rs.State = rshare.State
case "mount_point":
rs.MountPoint = rshare.MountPoint
case "hidden":
continue
default:
return nil, errtypes.NotSupported("updating " + fieldMask.Paths[i] + " is not supported")
}
}
uid := user.GetId()
if user.GetId().GetType() == userpb.UserType_USER_TYPE_SERVICE {
uid = forUser
}
err = m.persistReceivedShare(ctx, uid, rs)
if err != nil {
return nil, err
}
return rs, nil
}
func (m *Manager) persistReceivedShare(ctx context.Context, userID *userpb.UserId, rs *collaboration.ReceivedShare) error {
err := m.persistShare(ctx, rs.Share)
if err != nil {
return err
}
meta := ReceivedShareMetadata{
State: rs.State,
MountPoint: rs.MountPoint,
}
data, err := json.Marshal(meta)
if err != nil {
return err
}
fn, err := metadataFilename(rs.Share, userID)
if err != nil {
return err
}
return m.storage.SimpleUpload(ctx, fn, data)
}
func (m *Manager) downloadMetadata(ctx context.Context, share *collaboration.Share) (ReceivedShareMetadata, error) {
user, ok := ctxpkg.ContextGetUser(ctx)
if !ok {
return ReceivedShareMetadata{}, errtypes.UserRequired("error getting user from context")
}
metadataFn, err := metadataFilename(share, user.Id)
if err != nil {
return ReceivedShareMetadata{}, err
}
data, err := m.storage.SimpleDownload(ctx, metadataFn)
if err != nil {
return ReceivedShareMetadata{}, err
}
metadata := ReceivedShareMetadata{}
err = json.Unmarshal(data, &metadata)
return metadata, err
}
func (m *Manager) getShareByID(ctx context.Context, id string) (*collaboration.Share, error) {
data, err := m.storage.SimpleDownload(ctx, shareFilename(id))
if err != nil {
return nil, err
}
userShare := &collaboration.Share{
Grantee: &provider.Grantee{Id: &provider.Grantee_UserId{}},
}
err = json.Unmarshal(data, userShare)
if err == nil && userShare.Grantee.GetUserId() != nil {
userShare.ResourceId = storagespace.UpdateLegacyResourceID(userShare.GetResourceId())
return userShare, nil
}
groupShare := &collaboration.Share{
Grantee: &provider.Grantee{Id: &provider.Grantee_GroupId{}},
}
err = json.Unmarshal(data, groupShare) // try to unmarshal to a group share if the user share unmarshalling failed
if err == nil && groupShare.Grantee.GetGroupId() != nil {
groupShare.ResourceId = storagespace.UpdateLegacyResourceID(groupShare.GetResourceId())
return groupShare, nil
}
return nil, errtypes.InternalError("failed to unmarshal share data")
}
func (m *Manager) getShareByKey(ctx context.Context, key *collaboration.ShareKey) (*collaboration.Share, error) {
ownerIds, err := m.indexer.FindBy(&collaboration.Share{},
indexer.NewField("OwnerId", userIDToIndex(key.Owner)),
)
if err != nil {
return nil, err
}
granteeIndex, err := granteeToIndex(key.Grantee)
if err != nil {
return nil, err
}
granteeIds, err := m.indexer.FindBy(&collaboration.Share{},
indexer.NewField("GranteeId", granteeIndex),
)
if err != nil {
return nil, err
}
ids := intersectSlices(ownerIds, granteeIds)
for _, id := range ids {
share, err := m.getShareByID(ctx, id)
if err != nil {
return nil, err
}
if utils.ResourceIDEqual(share.ResourceId, key.ResourceId) {
return share, nil
}
}
return nil, errtypes.NotFound("share not found")
}
func shareFilename(id string) string {
return path.Join("shares", id)
}
func metadataFilename(s *collaboration.Share, g interface{}) (string, error) {
var granteePart string
switch v := g.(type) {
case *userpb.UserId:
granteePart = url.QueryEscape("user:" + v.Idp + ":" + v.OpaqueId)
case *provider.Grantee:
var err error
granteePart, err = granteeToIndex(v)
if err != nil {
return "", err
}
}
return path.Join("metadata", s.Id.OpaqueId, granteePart), nil
}
func indexOwnerFunc(v interface{}) (string, error) {
share, ok := v.(*collaboration.Share)
if !ok {
return "", fmt.Errorf("given entity is not a share")
}
return userIDToIndex(share.Owner), nil
}
func indexCreatorFunc(v interface{}) (string, error) {
share, ok := v.(*collaboration.Share)
if !ok {
return "", fmt.Errorf("given entity is not a share")
}
return userIDToIndex(share.Creator), nil
}
func userIDToIndex(id *userpb.UserId) string {
return url.QueryEscape(id.Idp + ":" + id.OpaqueId)
}
func indexGranteeFunc(v interface{}) (string, error) {
share, ok := v.(*collaboration.Share)
if !ok {
return "", fmt.Errorf("given entity is not a share")
}
return granteeToIndex(share.Grantee)
}
func indexResourceIDFunc(v interface{}) (string, error) {
share, ok := v.(*collaboration.Share)
if !ok {
return "", fmt.Errorf("given entity is not a share")
}
return resourceIDToIndex(share.ResourceId), nil
}
func resourceIDToIndex(id *provider.ResourceId) string {
return strings.Join([]string{id.SpaceId, id.OpaqueId}, "!")
}
func granteeToIndex(grantee *provider.Grantee) (string, error) {
switch {
case grantee.Type == provider.GranteeType_GRANTEE_TYPE_USER:
return url.QueryEscape("user:" + grantee.GetUserId().Idp + ":" + grantee.GetUserId().OpaqueId), nil
case grantee.Type == provider.GranteeType_GRANTEE_TYPE_GROUP:
return url.QueryEscape("group:" + grantee.GetGroupId().OpaqueId), nil
default:
return "", fmt.Errorf("unknown grantee type")
}
}
// indexToGrantee tries to unparse a grantee in a metadata dir
// unfortunately, it is just concatenated by :, causing nasty corner cases
func indexToGrantee(name string) (*provider.Grantee, error) {
unescaped, err := url.QueryUnescape(name)
if err != nil {
return nil, err
}
parts := strings.SplitN(unescaped, ":", 2)
if len(parts) != 2 {
return nil, fmt.Errorf("invalid grantee %s", unescaped)
}
switch parts[0] {
case "user":
lastInd := strings.LastIndex(parts[1], ":")
return &provider.Grantee{
Type: provider.GranteeType_GRANTEE_TYPE_USER,
Id: &provider.Grantee_UserId{
UserId: &userpb.UserId{
Idp: parts[1][:lastInd],
OpaqueId: parts[1][lastInd+1:],
},
},
}, nil
case "group":
return &provider.Grantee{
Type: provider.GranteeType_GRANTEE_TYPE_GROUP,
Id: &provider.Grantee_GroupId{
GroupId: &groupv1beta1.GroupId{
OpaqueId: parts[1],
},
},
}, nil
default:
return nil, fmt.Errorf("invalid grantee %s", unescaped)
}
}
func intersectSlices(a, b []string) []string {
aMap := map[string]bool{}
for _, s := range a {
aMap[s] = true
}
result := []string{}
for _, s := range b {
if _, ok := aMap[s]; ok {
result = append(result, s)
}
}
return result
}
+617
View File
@@ -0,0 +1,617 @@
// Copyright 2018-2021 CERN
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
//
// In applying this license, CERN does not waive the privileges and immunities
// granted to it by virtue of its status as an Intergovernmental Organization
// or submit itself to any jurisdiction.
package json
import (
"context"
"encoding/json"
"io"
"io/fs"
"os"
"strings"
"sync"
"time"
userv1beta1 "github.com/cs3org/go-cs3apis/cs3/identity/user/v1beta1"
rpcv1beta1 "github.com/cs3org/go-cs3apis/cs3/rpc/v1beta1"
collaboration "github.com/cs3org/go-cs3apis/cs3/sharing/collaboration/v1beta1"
provider "github.com/cs3org/go-cs3apis/cs3/storage/provider/v1beta1"
typespb "github.com/cs3org/go-cs3apis/cs3/types/v1beta1"
"github.com/golang/protobuf/proto" // nolint:staticcheck // we need the legacy package to convert V1 to V2 messages
"github.com/google/uuid"
"github.com/mitchellh/mapstructure"
"github.com/opencloud-eu/reva/v2/pkg/appctx"
ctxpkg "github.com/opencloud-eu/reva/v2/pkg/ctx"
"github.com/opencloud-eu/reva/v2/pkg/errtypes"
"github.com/opencloud-eu/reva/v2/pkg/rgrpc/todo/pool"
"github.com/opencloud-eu/reva/v2/pkg/share"
"github.com/pkg/errors"
"google.golang.org/genproto/protobuf/field_mask"
"google.golang.org/protobuf/encoding/prototext"
"github.com/opencloud-eu/reva/v2/pkg/share/manager/registry"
"github.com/opencloud-eu/reva/v2/pkg/utils"
)
func init() {
registry.Register("json", New)
}
// New returns a new mgr.
func New(m map[string]interface{}) (share.Manager, error) {
c, err := parseConfig(m)
if err != nil {
err = errors.Wrap(err, "error creating a new manager")
return nil, err
}
if c.GatewayAddr == "" {
return nil, errors.New("share manager config is missing gateway address")
}
c.init()
// load or create file
model, err := loadOrCreate(c.File)
if err != nil {
err = errors.Wrap(err, "error loading the file containing the shares")
return nil, err
}
return &mgr{
c: c,
model: model,
}, nil
}
func loadOrCreate(file string) (*shareModel, error) {
if info, err := os.Stat(file); errors.Is(err, fs.ErrNotExist) || info.Size() == 0 {
if err := os.WriteFile(file, []byte("{}"), 0700); err != nil {
err = errors.Wrap(err, "error opening/creating the file: "+file)
return nil, err
}
}
fd, err := os.OpenFile(file, os.O_CREATE, 0644)
if err != nil {
err = errors.Wrap(err, "error opening/creating the file: "+file)
return nil, err
}
defer fd.Close()
data, err := io.ReadAll(fd)
if err != nil {
err = errors.Wrap(err, "error reading the data")
return nil, err
}
j := &jsonEncoding{}
if err := json.Unmarshal(data, j); err != nil {
err = errors.Wrap(err, "error decoding data from json")
return nil, err
}
m := &shareModel{State: j.State, MountPoint: j.MountPoint}
for _, s := range j.Shares {
var decShare collaboration.Share
if err = utils.UnmarshalJSONToProtoV1([]byte(s), &decShare); err != nil {
return nil, errors.Wrap(err, "error decoding share from json")
}
m.Shares = append(m.Shares, &decShare)
}
if m.State == nil {
m.State = map[string]map[string]collaboration.ShareState{}
}
if m.MountPoint == nil {
m.MountPoint = map[string]map[string]*provider.Reference{}
}
m.file = file
return m, nil
}
type shareModel struct {
file string
State map[string]map[string]collaboration.ShareState `json:"state"` // map[username]map[share_id]ShareState
MountPoint map[string]map[string]*provider.Reference `json:"mount_point"` // map[username]map[share_id]MountPoint
Shares []*collaboration.Share `json:"shares"`
}
type jsonEncoding struct {
State map[string]map[string]collaboration.ShareState `json:"state"` // map[username]map[share_id]ShareState
MountPoint map[string]map[string]*provider.Reference `json:"mount_point"` // map[username]map[share_id]MountPoint
Shares []string `json:"shares"`
}
func (m *shareModel) Save() error {
j := &jsonEncoding{State: m.State, MountPoint: m.MountPoint}
for _, s := range m.Shares {
encShare, err := utils.MarshalProtoV1ToJSON(s)
if err != nil {
return errors.Wrap(err, "error encoding to json")
}
j.Shares = append(j.Shares, string(encShare))
}
data, err := json.Marshal(j)
if err != nil {
err = errors.Wrap(err, "error encoding to json")
return err
}
if err := os.WriteFile(m.file, data, 0644); err != nil {
err = errors.Wrap(err, "error writing to file: "+m.file)
return err
}
return nil
}
type mgr struct {
c *config
sync.Mutex // concurrent access to the file
model *shareModel
}
type config struct {
File string `mapstructure:"file"`
GatewayAddr string `mapstructure:"gateway_addr"`
}
func (c *config) init() {
if c.File == "" {
c.File = "/var/tmp/reva/shares.json"
}
}
func parseConfig(m map[string]interface{}) (*config, error) {
c := &config{}
if err := mapstructure.Decode(m, c); err != nil {
return nil, err
}
return c, nil
}
// Dump exports shares and received shares to channels (e.g. during migration)
func (m *mgr) Dump(ctx context.Context, shareChan chan<- *collaboration.Share, receivedShareChan chan<- share.ReceivedShareWithUser) error {
log := appctx.GetLogger(ctx)
for _, s := range m.model.Shares {
shareChan <- s
}
for userIDString, states := range m.model.State {
userMountPoints := m.model.MountPoint[userIDString]
id := &userv1beta1.UserId{}
mV2 := proto.MessageV2(id)
if err := prototext.Unmarshal([]byte(userIDString), mV2); err != nil {
log.Error().Err(err).Msg("error unmarshalling the user id")
continue
}
for shareIDString, state := range states {
sid := &collaboration.ShareId{}
mV2 := proto.MessageV2(sid)
if err := prototext.Unmarshal([]byte(shareIDString), mV2); err != nil {
log.Error().Err(err).Msg("error unmarshalling the user id")
continue
}
var s *collaboration.Share
for _, is := range m.model.Shares {
if is.Id.OpaqueId == sid.OpaqueId {
s = is
break
}
}
if s == nil {
log.Warn().Str("share id", sid.OpaqueId).Msg("Share not found")
continue
}
var mp *provider.Reference
if userMountPoints != nil {
mp = userMountPoints[shareIDString]
}
receivedShareChan <- share.ReceivedShareWithUser{
UserID: id,
ReceivedShare: &collaboration.ReceivedShare{
Share: s,
State: state,
MountPoint: mp,
},
}
}
}
return nil
}
func (m *mgr) Share(ctx context.Context, md *provider.ResourceInfo, g *collaboration.ShareGrant) (*collaboration.Share, error) {
id := uuid.NewString()
user := ctxpkg.ContextMustGetUser(ctx)
now := time.Now().UnixNano()
ts := &typespb.Timestamp{
Seconds: uint64(now / int64(time.Second)),
Nanos: uint32(now % int64(time.Second)),
}
// do not allow share to myself or the owner if share is for a user
// TODO(labkode): should not this be caught already at the gw level?
if g.Grantee.Type == provider.GranteeType_GRANTEE_TYPE_USER &&
(utils.UserEqual(g.Grantee.GetUserId(), user.Id) || utils.UserEqual(g.Grantee.GetUserId(), md.Owner)) {
return nil, errtypes.BadRequest("json: owner/creator and grantee are the same")
}
// check if share already exists.
key := &collaboration.ShareKey{
Owner: md.Owner,
ResourceId: md.Id,
Grantee: g.Grantee,
}
m.Lock()
defer m.Unlock()
_, _, err := m.getByKey(key)
if err == nil {
// share already exists
return nil, errtypes.AlreadyExists(key.String())
}
s := &collaboration.Share{
Id: &collaboration.ShareId{
OpaqueId: id,
},
ResourceId: md.Id,
Permissions: g.Permissions,
Grantee: g.Grantee,
Owner: md.Owner,
Creator: user.Id,
Ctime: ts,
Mtime: ts,
}
m.model.Shares = append(m.model.Shares, s)
if err := m.model.Save(); err != nil {
err = errors.Wrap(err, "error saving model")
return nil, err
}
return s, nil
}
// getByID must be called in a lock-controlled block.
func (m *mgr) getByID(id *collaboration.ShareId) (int, *collaboration.Share, error) {
for i, s := range m.model.Shares {
if s.GetId().OpaqueId == id.OpaqueId {
return i, s, nil
}
}
return -1, nil, errtypes.NotFound(id.String())
}
// getByKey must be called in a lock-controlled block.
func (m *mgr) getByKey(key *collaboration.ShareKey) (int, *collaboration.Share, error) {
for i, s := range m.model.Shares {
if (utils.UserEqual(key.Owner, s.Owner) || utils.UserEqual(key.Owner, s.Creator)) &&
utils.ResourceIDEqual(key.ResourceId, s.ResourceId) && utils.GranteeEqual(key.Grantee, s.Grantee) {
return i, s, nil
}
}
return -1, nil, errtypes.NotFound(key.String())
}
// get must be called in a lock-controlled block.
func (m *mgr) get(ref *collaboration.ShareReference) (idx int, s *collaboration.Share, err error) {
switch {
case ref.GetId() != nil:
idx, s, err = m.getByID(ref.GetId())
case ref.GetKey() != nil:
idx, s, err = m.getByKey(ref.GetKey())
default:
err = errtypes.NotFound(ref.String())
}
return
}
func (m *mgr) GetShare(ctx context.Context, ref *collaboration.ShareReference) (*collaboration.Share, error) {
m.Lock()
defer m.Unlock()
_, s, err := m.get(ref)
if err != nil {
return nil, err
}
// check if we are the owner or the grantee
user := ctxpkg.ContextMustGetUser(ctx)
if share.IsCreatedByUser(s, user) || share.IsGrantedToUser(s, user) {
return s, nil
}
// we return not found to not disclose information
return nil, errtypes.NotFound(ref.String())
}
func (m *mgr) Unshare(ctx context.Context, ref *collaboration.ShareReference) error {
m.Lock()
defer m.Unlock()
user := ctxpkg.ContextMustGetUser(ctx)
idx, s, err := m.get(ref)
if err != nil {
return err
}
if !share.IsCreatedByUser(s, user) {
return errtypes.NotFound(ref.String())
}
last := len(m.model.Shares) - 1
m.model.Shares[idx] = m.model.Shares[last]
// explicitly nil the reference to prevent memory leaks
// https://github.com/golang/go/wiki/SliceTricks#delete-without-preserving-order
m.model.Shares[last] = nil
m.model.Shares = m.model.Shares[:last]
if err := m.model.Save(); err != nil {
err = errors.Wrap(err, "error saving model")
return err
}
return nil
}
func (m *mgr) UpdateShare(ctx context.Context, ref *collaboration.ShareReference, p *collaboration.SharePermissions, updated *collaboration.Share, fieldMask *field_mask.FieldMask) (*collaboration.Share, error) {
m.Lock()
defer m.Unlock()
var (
idx int
toUpdate *collaboration.Share
)
if ref != nil {
var err error
idx, toUpdate, err = m.get(ref)
if err != nil {
return nil, err
}
} else if updated != nil {
var err error
idx, toUpdate, err = m.getByID(updated.Id)
if err != nil {
return nil, err
}
}
if fieldMask != nil {
for i := range fieldMask.Paths {
switch fieldMask.Paths[i] {
case "permissions":
m.model.Shares[idx].Permissions = updated.Permissions
case "expiration":
m.model.Shares[idx].Expiration = updated.Expiration
case "hidden":
continue
default:
return nil, errtypes.NotSupported("updating " + fieldMask.Paths[i] + " is not supported")
}
}
}
user := ctxpkg.ContextMustGetUser(ctx)
if !share.IsCreatedByUser(toUpdate, user) {
return nil, errtypes.NotFound(ref.String())
}
now := time.Now().UnixNano()
if p != nil {
m.model.Shares[idx].Permissions = p
}
m.model.Shares[idx].Mtime = &typespb.Timestamp{
Seconds: uint64(now / int64(time.Second)),
Nanos: uint32(now % int64(time.Second)),
}
if err := m.model.Save(); err != nil {
err = errors.Wrap(err, "error saving model")
return nil, err
}
return m.model.Shares[idx], nil
}
func (m *mgr) ListShares(ctx context.Context, filters []*collaboration.Filter) ([]*collaboration.Share, error) {
m.Lock()
defer m.Unlock()
log := appctx.GetLogger(ctx)
user := ctxpkg.ContextMustGetUser(ctx)
client, err := pool.GetGatewayServiceClient(m.c.GatewayAddr)
if err != nil {
return nil, errors.Wrap(err, "failed to list shares")
}
cache := make(map[string]struct{})
var ss []*collaboration.Share
for _, s := range m.model.Shares {
if share.MatchesFilters(s, filters) {
// Only add the share if the share was created by the user or if
// the user has ListGrants permissions on the shared resource.
// The ListGrants check is necessary when a space member wants
// to list shares in a space.
// We are using a cache here so that we don't have to stat a
// resource multiple times.
key := strings.Join([]string{s.ResourceId.StorageId, s.ResourceId.OpaqueId}, "!")
if _, hit := cache[key]; !hit && !share.IsCreatedByUser(s, user) {
sRes, err := client.Stat(ctx, &provider.StatRequest{Ref: &provider.Reference{ResourceId: s.ResourceId}})
if err != nil || sRes.Status.Code != rpcv1beta1.Code_CODE_OK {
log.Error().
Err(err).
Interface("status", sRes.Status).
Interface("resource_id", s.ResourceId).
Msg("ListShares: could not stat resource")
continue
}
if !sRes.Info.PermissionSet.ListGrants {
continue
}
cache[key] = struct{}{}
}
ss = append(ss, s)
}
}
return ss, nil
}
// we list the shares that are targeted to the user in context or to the user groups.
func (m *mgr) ListReceivedShares(ctx context.Context, filters []*collaboration.Filter, forUser *userv1beta1.UserId) ([]*collaboration.ReceivedShare, error) {
m.Lock()
defer m.Unlock()
user := ctxpkg.ContextMustGetUser(ctx)
if user.GetId().GetType() == userv1beta1.UserType_USER_TYPE_SERVICE {
gwc, err := pool.GetGatewayServiceClient(m.c.GatewayAddr)
if err != nil {
return nil, errors.Wrap(err, "failed to list shares")
}
u, err := utils.GetUser(forUser, gwc)
if err != nil {
return nil, errtypes.BadRequest("user not found")
}
user = u
}
mem := make(map[string]int)
var rss []*collaboration.ReceivedShare
for _, s := range m.model.Shares {
if !share.IsCreatedByUser(s, user) &&
share.IsGrantedToUser(s, user) &&
share.MatchesFilters(s, filters) {
rs := m.convert(user.Id, s)
idx, seen := mem[s.ResourceId.OpaqueId]
if !seen {
rss = append(rss, rs)
mem[s.ResourceId.OpaqueId] = len(rss) - 1
continue
}
// When we arrive here there was already a share for this resource.
// if there is a mix-up of shares of type group and shares of type user we need to deduplicate them, since it points
// to the same resource. Leave the more explicit and hide the less explicit. In this case we hide the group shares
// and return the user share to the user.
other := rss[idx]
if other.Share.Grantee.Type == provider.GranteeType_GRANTEE_TYPE_GROUP && s.Grantee.Type == provider.GranteeType_GRANTEE_TYPE_USER {
if other.State == rs.State {
rss[idx] = rs
} else {
rss = append(rss, rs)
}
}
}
}
return rss, nil
}
// convert must be called in a lock-controlled block.
func (m *mgr) convert(currentUser *userv1beta1.UserId, s *collaboration.Share) *collaboration.ReceivedShare {
rs := &collaboration.ReceivedShare{
Share: s,
State: collaboration.ShareState_SHARE_STATE_PENDING,
}
if v, ok := m.model.State[currentUser.String()]; ok {
if state, ok := v[s.Id.String()]; ok {
rs.State = state
}
}
if v, ok := m.model.MountPoint[currentUser.String()]; ok {
if mp, ok := v[s.Id.String()]; ok {
rs.MountPoint = mp
}
}
return rs
}
func (m *mgr) GetReceivedShare(ctx context.Context, ref *collaboration.ShareReference) (*collaboration.ReceivedShare, error) {
return m.getReceived(ctx, ref)
}
func (m *mgr) getReceived(ctx context.Context, ref *collaboration.ShareReference) (*collaboration.ReceivedShare, error) {
m.Lock()
defer m.Unlock()
_, s, err := m.get(ref)
if err != nil {
return nil, err
}
user := ctxpkg.ContextMustGetUser(ctx)
if user.GetId().GetType() != userv1beta1.UserType_USER_TYPE_SERVICE && !share.IsGrantedToUser(s, user) {
return nil, errtypes.NotFound(ref.String())
}
return m.convert(user.Id, s), nil
}
func (m *mgr) UpdateReceivedShare(ctx context.Context, receivedShare *collaboration.ReceivedShare, fieldMask *field_mask.FieldMask, forUser *userv1beta1.UserId) (*collaboration.ReceivedShare, error) {
rs, err := m.getReceived(ctx, &collaboration.ShareReference{Spec: &collaboration.ShareReference_Id{Id: receivedShare.Share.Id}})
if err != nil {
return nil, err
}
m.Lock()
defer m.Unlock()
for i := range fieldMask.Paths {
switch fieldMask.Paths[i] {
case "state":
rs.State = receivedShare.State
case "mount_point":
rs.MountPoint = receivedShare.MountPoint
default:
return nil, errtypes.NotSupported("updating " + fieldMask.Paths[i] + " is not supported")
}
}
u := ctxpkg.ContextMustGetUser(ctx)
uid := u.GetId().String()
if u.GetId().GetType() == userv1beta1.UserType_USER_TYPE_SERVICE {
uid = forUser.String()
}
// Persist state
if v, ok := m.model.State[uid]; ok {
v[rs.Share.Id.String()] = rs.State
m.model.State[uid] = v
} else {
a := map[string]collaboration.ShareState{
rs.Share.Id.String(): rs.State,
}
m.model.State[uid] = a
}
// Persist mount point
if v, ok := m.model.MountPoint[uid]; ok {
v[rs.Share.Id.String()] = rs.MountPoint
m.model.MountPoint[uid] = v
} else {
a := map[string]*provider.Reference{
rs.Share.Id.String(): rs.MountPoint,
}
m.model.MountPoint[uid] = a
}
if err := m.model.Save(); err != nil {
err = errors.Wrap(err, "error saving model")
return nil, err
}
return rs, nil
}
File diff suppressed because it is too large Load Diff
@@ -0,0 +1,543 @@
// Copyright 2018-2021 CERN
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
//
// In applying this license, CERN does not waive the privileges and immunities
// granted to it by virtue of its status as an Intergovernmental Organization
// or submit itself to any jurisdiction.
package providercache
import (
"context"
"encoding/json"
"fmt"
"os"
"path"
"path/filepath"
"strings"
"sync"
"time"
collaboration "github.com/cs3org/go-cs3apis/cs3/sharing/collaboration/v1beta1"
provider "github.com/cs3org/go-cs3apis/cs3/storage/provider/v1beta1"
"github.com/opencloud-eu/reva/v2/pkg/appctx"
"github.com/opencloud-eu/reva/v2/pkg/errtypes"
"github.com/opencloud-eu/reva/v2/pkg/storage/utils/decomposedfs/mtimesyncedcache"
"github.com/opencloud-eu/reva/v2/pkg/storage/utils/metadata"
"go.opentelemetry.io/otel"
"go.opentelemetry.io/otel/attribute"
"go.opentelemetry.io/otel/codes"
"go.opentelemetry.io/otel/trace"
"golang.org/x/exp/maps"
)
var tracer trace.Tracer
func init() {
tracer = otel.Tracer("github.com/opencloud-eu/reva/v2/pkg/share/manager/jsoncs3/providercache")
}
// Cache holds share information structured by provider and space
type Cache struct {
lockMap sync.Map
Providers mtimesyncedcache.Map[string, *Spaces]
storage metadata.Storage
ttl time.Duration
}
// Spaces holds the share information for provider
type Spaces struct {
Spaces mtimesyncedcache.Map[string, *Shares]
}
// Shares holds the share information of one space
type Shares struct {
Shares map[string]*collaboration.Share
Etag string
}
// UnmarshalJSON overrides the default unmarshaling
// Shares are tricky to unmarshal because they contain an interface (Grantee) which makes the json Unmarshal bail out
// To work around that problem we unmarshal into json.RawMessage in a first step and then try to manually unmarshal
// into the specific types in a second step.
func (s *Shares) UnmarshalJSON(data []byte) error {
tmp := struct {
Shares map[string]json.RawMessage
}{}
err := json.Unmarshal(data, &tmp)
if err != nil {
return err
}
s.Shares = make(map[string]*collaboration.Share, len(tmp.Shares))
for id, genericShare := range tmp.Shares {
userShare := &collaboration.Share{
Grantee: &provider.Grantee{Id: &provider.Grantee_UserId{}},
}
err = json.Unmarshal(genericShare, userShare) // is this a user share?
if err == nil && userShare.Grantee.Type == provider.GranteeType_GRANTEE_TYPE_USER {
s.Shares[id] = userShare
continue
}
groupShare := &collaboration.Share{
Grantee: &provider.Grantee{Id: &provider.Grantee_GroupId{}},
}
err = json.Unmarshal(genericShare, groupShare) // is this a group share?
if err == nil && groupShare.Grantee.Type == provider.GranteeType_GRANTEE_TYPE_GROUP {
s.Shares[id] = groupShare
continue
}
invalidShare := &collaboration.Share{}
err = json.Unmarshal(genericShare, invalidShare) // invalid
if err == nil {
s.Shares[id] = invalidShare
continue
}
return err
}
return nil
}
// LockSpace locks the cache for a given space and returns an unlock function
func (c *Cache) LockSpace(spaceID string) func() {
v, _ := c.lockMap.LoadOrStore(spaceID, &sync.Mutex{})
lock := v.(*sync.Mutex)
lock.Lock()
return func() { lock.Unlock() }
}
// New returns a new Cache instance
func New(s metadata.Storage, ttl time.Duration) Cache {
return Cache{
Providers: mtimesyncedcache.Map[string, *Spaces]{},
storage: s,
ttl: ttl,
lockMap: sync.Map{},
}
}
func (c *Cache) isSpaceCached(storageID, spaceID string) bool {
spaces, ok := c.Providers.Load(storageID)
if !ok {
return false
}
_, ok = spaces.Spaces.Load(spaceID)
return ok
}
// Add adds a share to the cache
func (c *Cache) Add(ctx context.Context, storageID, spaceID, shareID string, share *collaboration.Share) error {
ctx, span := tracer.Start(ctx, "Add")
defer span.End()
span.SetAttributes(attribute.String("cs3.storageid", storageID), attribute.String("cs3.spaceid", spaceID), attribute.String("cs3.shareid", shareID))
switch {
case storageID == "":
return fmt.Errorf("missing storage id")
case spaceID == "":
return fmt.Errorf("missing space id")
case shareID == "":
return fmt.Errorf("missing share id")
}
unlock := c.LockSpace(spaceID)
defer unlock()
span.AddEvent("got lock")
var err error
if !c.isSpaceCached(storageID, spaceID) {
err = c.syncWithLock(ctx, storageID, spaceID)
if err != nil {
return err
}
}
log := appctx.GetLogger(ctx).With().
Str("hostname", os.Getenv("HOSTNAME")).
Str("storageID", storageID).
Str("spaceID", spaceID).
Str("shareID", shareID).Logger()
persistFunc := func() error {
spaces, _ := c.Providers.Load(storageID)
space, _ := spaces.Spaces.Load(spaceID)
log.Info().Interface("shares", maps.Keys(space.Shares)).Str("New share", shareID).Msg("Adding share to space")
space.Shares[shareID] = share
return c.Persist(ctx, storageID, spaceID)
}
for retries := 100; retries > 0; retries-- {
err = persistFunc()
switch err.(type) {
case nil:
span.SetStatus(codes.Ok, "")
return nil
case errtypes.Aborted:
log.Debug().Msg("aborted when persisting added provider share: etag changed. retrying...")
// this is the expected status code from the server when the if-match etag check fails
// continue with sync below
case errtypes.PreconditionFailed:
log.Debug().Msg("precondition failed when persisting added provider share: etag changed. retrying...")
// actually, this is the wrong status code and we treat it like errtypes.Aborted because of inconsistencies on the server side
// continue with sync below
case errtypes.AlreadyExists:
log.Debug().Msg("already exists when persisting added provider share. retrying...")
// CS3 uses an already exists error instead of precondition failed when using an If-None-Match=* header / IfExists flag in the InitiateFileUpload call.
// Thas happens when the cache thinks there is no file.
// continue with sync below
default:
span.SetStatus(codes.Error, fmt.Sprintf("persisting added provider share failed. giving up: %s", err.Error()))
log.Error().Err(err).Msg("persisting added provider share failed")
return err
}
if err := c.syncWithLock(ctx, storageID, spaceID); err != nil {
span.RecordError(err)
span.SetStatus(codes.Error, err.Error())
log.Error().Err(err).Msg("persisting added provider share failed. giving up.")
return err
}
}
return err
}
// Remove removes a share from the cache
func (c *Cache) Remove(ctx context.Context, storageID, spaceID, shareID string) error {
ctx, span := tracer.Start(ctx, "Remove")
defer span.End()
span.SetAttributes(attribute.String("cs3.storageid", storageID), attribute.String("cs3.spaceid", spaceID), attribute.String("cs3.shareid", shareID))
unlock := c.LockSpace(spaceID)
defer unlock()
span.AddEvent("got lock")
if !c.isSpaceCached(storageID, spaceID) {
err := c.syncWithLock(ctx, storageID, spaceID)
if err != nil {
return err
}
}
persistFunc := func() error {
spaces, ok := c.Providers.Load(storageID)
if !ok {
return nil
}
space, _ := spaces.Spaces.Load(spaceID)
if !ok {
return nil
}
delete(space.Shares, shareID)
return c.Persist(ctx, storageID, spaceID)
}
log := appctx.GetLogger(ctx).With().
Str("hostname", os.Getenv("HOSTNAME")).
Str("storageID", storageID).
Str("spaceID", spaceID).
Str("shareID", shareID).Logger()
var err error
for retries := 100; retries > 0; retries-- {
err = persistFunc()
switch err.(type) {
case nil:
span.SetStatus(codes.Ok, "")
return nil
case errtypes.Aborted:
log.Debug().Msg("aborted when persisting removed provider share: etag changed. retrying...")
// this is the expected status code from the server when the if-match etag check fails
// continue with sync below
case errtypes.PreconditionFailed:
log.Debug().Msg("precondition failed when persisting removed provider share: etag changed. retrying...")
// actually, this is the wrong status code and we treat it like errtypes.Aborted because of inconsistencies on the server side
// continue with sync below
default:
span.SetStatus(codes.Error, fmt.Sprintf("persisting removed provider share failed. giving up: %s", err.Error()))
log.Error().Err(err).Msg("persisting removed provider share failed")
return err
}
if err := c.syncWithLock(ctx, storageID, spaceID); err != nil {
span.RecordError(err)
span.SetStatus(codes.Error, err.Error())
log.Error().Err(err).Msg("persisting removed provider share failed. giving up.")
return err
}
}
return err
}
// Get returns one entry from the cache
func (c *Cache) Get(ctx context.Context, storageID, spaceID, shareID string, skipSync bool) (*collaboration.Share, error) {
ctx, span := tracer.Start(ctx, "Get")
defer span.End()
span.SetAttributes(attribute.String("cs3.storageid", storageID), attribute.String("cs3.spaceid", spaceID), attribute.String("cs3.shareid", shareID))
unlock := c.LockSpace(spaceID)
defer unlock()
span.AddEvent("got lock")
if !skipSync {
// sync cache, maybe our data is outdated
err := c.syncWithLock(ctx, storageID, spaceID)
if err != nil {
return nil, err
}
}
spaces, ok := c.Providers.Load(storageID)
if !ok {
return nil, nil
}
space, ok := spaces.Spaces.Load(spaceID)
if !ok {
return nil, nil
}
return space.Shares[shareID], nil
}
// All returns all entries in the storage
func (c *Cache) All(ctx context.Context) (*mtimesyncedcache.Map[string, *Spaces], error) {
ctx, span := tracer.Start(ctx, "All")
defer span.End()
providers, err := c.storage.ListDir(ctx, "/storages")
if err != nil {
return nil, err
}
for _, provider := range providers {
storageID := provider.Name
spaces, err := c.storage.ListDir(ctx, path.Join("/storages", storageID))
if err != nil {
return nil, err
}
for _, space := range spaces {
spaceID := strings.TrimSuffix(space.Name, ".json")
unlock := c.LockSpace(spaceID)
span.AddEvent("got lock for space " + spaceID)
if err := c.syncWithLock(ctx, storageID, spaceID); err != nil {
return nil, err
}
unlock()
}
}
return &c.Providers, nil
}
// ListSpace returns the list of shares in a given space
func (c *Cache) ListSpace(ctx context.Context, storageID, spaceID string) (*Shares, error) {
ctx, span := tracer.Start(ctx, "ListSpace")
defer span.End()
span.SetAttributes(attribute.String("cs3.storageid", storageID), attribute.String("cs3.spaceid", spaceID))
unlock := c.LockSpace(spaceID)
defer unlock()
span.AddEvent("got lock")
// sync cache, maybe our data is outdated
err := c.syncWithLock(ctx, storageID, spaceID)
if err != nil {
return nil, err
}
spaces, ok := c.Providers.Load(storageID)
if !ok {
return &Shares{}, nil
}
space, ok := spaces.Spaces.Load(spaceID)
if !ok {
return &Shares{}, nil
}
shares := &Shares{
Shares: maps.Clone(space.Shares),
Etag: space.Etag,
}
return shares, nil
}
// Persist persists the data of one space
func (c *Cache) Persist(ctx context.Context, storageID, spaceID string) error {
ctx, span := tracer.Start(ctx, "Persist")
defer span.End()
span.SetAttributes(attribute.String("cs3.storageid", storageID), attribute.String("cs3.spaceid", spaceID))
spaces, ok := c.Providers.Load(storageID)
if !ok {
span.AddEvent("nothing to persist")
span.SetStatus(codes.Ok, "")
return nil
}
space, ok := spaces.Spaces.Load(spaceID)
if !ok {
span.AddEvent("nothing to persist")
span.SetStatus(codes.Ok, "")
return nil
}
span.SetAttributes(attribute.String("BeforeEtag", space.Etag))
log := appctx.GetLogger(ctx).With().Str("storageID", storageID).Str("spaceID", spaceID).Logger()
log = log.With().Str("BeforeEtag", space.Etag).Logger()
createdBytes, err := json.Marshal(space)
if err != nil {
span.RecordError(err)
span.SetStatus(codes.Error, err.Error())
return err
}
jsonPath := spaceJSONPath(storageID, spaceID)
if err := c.storage.MakeDirIfNotExist(ctx, path.Dir(jsonPath)); err != nil {
span.RecordError(err)
span.SetStatus(codes.Error, err.Error())
return err
}
span.SetAttributes(attribute.String("etag", space.Etag))
ur := metadata.UploadRequest{
Path: jsonPath,
Content: createdBytes,
IfMatchEtag: space.Etag,
}
// when there is no etag in memory make sure the file has not been created on the server, see https://www.rfc-editor.org/rfc/rfc9110#field.if-match
// > If the field value is "*", the condition is false if the origin server has a current representation for the target resource.
if space.Etag == "" {
ur.IfNoneMatch = []string{"*"}
}
res, err := c.storage.Upload(ctx, ur)
if err != nil {
span.RecordError(err)
span.SetStatus(codes.Error, err.Error())
log.Debug().Err(err).Msg("persisting provider cache failed")
return err
}
space.Etag = res.Etag
span.SetStatus(codes.Ok, "")
shares := []string{}
for _, s := range space.Shares {
shares = append(shares, s.GetId().GetOpaqueId())
}
log.Debug().Str("AfterEtag", space.Etag).Interface("Shares", shares).Msg("persisted provider cache")
return nil
}
// PurgeSpace removes a space from the cache
func (c *Cache) PurgeSpace(ctx context.Context, storageID, spaceID string) error {
ctx, span := tracer.Start(ctx, "PurgeSpace")
defer span.End()
unlock := c.LockSpace(spaceID)
defer unlock()
span.AddEvent("got lock")
if !c.isSpaceCached(storageID, spaceID) {
err := c.syncWithLock(ctx, storageID, spaceID)
if err != nil {
return err
}
}
spaces, ok := c.Providers.Load(storageID)
if !ok {
return nil
}
newShares := &Shares{}
if space, ok := spaces.Spaces.Load(spaceID); ok {
newShares.Etag = space.Etag // keep the etag to allow overwriting the state on the server
}
spaces.Spaces.Store(spaceID, newShares)
return c.Persist(ctx, storageID, spaceID)
}
func (c *Cache) syncWithLock(ctx context.Context, storageID, spaceID string) error {
ctx, span := tracer.Start(ctx, "syncWithLock")
defer span.End()
c.initializeIfNeeded(storageID, spaceID)
spaces, _ := c.Providers.Load(storageID)
space, _ := spaces.Spaces.Load(spaceID)
span.SetAttributes(attribute.String("cs3.storageid", storageID), attribute.String("cs3.spaceid", spaceID), attribute.String("etag", space.Etag))
log := appctx.GetLogger(ctx).With().Str("storageID", storageID).Str("spaceID", spaceID).Str("etag", space.Etag).Str("hostname", os.Getenv("HOSTNAME")).Logger()
dlreq := metadata.DownloadRequest{
Path: spaceJSONPath(storageID, spaceID),
}
// when we know an etag, only download if it changed remotely
if space.Etag != "" {
dlreq.IfNoneMatch = []string{space.Etag}
}
dlres, err := c.storage.Download(ctx, dlreq)
switch err.(type) {
case nil:
span.AddEvent("updating local cache")
case errtypes.NotFound:
span.SetStatus(codes.Ok, "")
return nil
case errtypes.NotModified:
span.SetStatus(codes.Ok, "")
return nil
default:
span.RecordError(err)
span.SetStatus(codes.Error, "downloading provider cache failed")
return err
}
span.AddEvent("updating local cache")
newShares := &Shares{}
err = json.Unmarshal(dlres.Content, newShares)
if err != nil {
span.RecordError(err)
span.SetStatus(codes.Error, "unmarshaling provider cache failed")
log.Error().Err(err).Msg("unmarshaling provider cache failed")
return err
}
newShares.Etag = dlres.Etag
spaces.Spaces.Store(spaceID, newShares)
span.SetStatus(codes.Ok, "")
return nil
}
func (c *Cache) initializeIfNeeded(storageID, spaceID string) {
spaces, _ := c.Providers.LoadOrStore(storageID, &Spaces{
Spaces: mtimesyncedcache.Map[string, *Shares]{},
})
_, _ = spaces.Spaces.LoadOrStore(spaceID, &Shares{
Shares: map[string]*collaboration.Share{},
})
}
func spaceJSONPath(storageID, spaceID string) string {
return filepath.Join("/storages", storageID, spaceID+".json")
}
@@ -0,0 +1,392 @@
// Copyright 2018-2022 CERN
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
//
// In applying this license, CERN does not waive the privileges and immunities
// granted to it by virtue of its status as an Intergovernmental Organization
// or submit itself to any jurisdiction.
package receivedsharecache
import (
"context"
"encoding/json"
"fmt"
"os"
"path"
"path/filepath"
"sync"
"time"
collaboration "github.com/cs3org/go-cs3apis/cs3/sharing/collaboration/v1beta1"
provider "github.com/cs3org/go-cs3apis/cs3/storage/provider/v1beta1"
"github.com/opencloud-eu/reva/v2/pkg/appctx"
"github.com/opencloud-eu/reva/v2/pkg/errtypes"
"github.com/opencloud-eu/reva/v2/pkg/storage/utils/decomposedfs/mtimesyncedcache"
"github.com/opencloud-eu/reva/v2/pkg/storage/utils/metadata"
"go.opentelemetry.io/otel/attribute"
"go.opentelemetry.io/otel/codes"
)
// name is the Tracer name used to identify this instrumentation library.
const tracerName = "receivedsharecache"
// Cache stores the list of received shares and their states
// It functions as an in-memory cache with a persistence layer
// The storage is sharded by user
type Cache struct {
lockMap sync.Map
ReceivedSpaces mtimesyncedcache.Map[string, *Spaces]
storage metadata.Storage
ttl time.Duration
}
// Spaces holds the received shares of one user per space
type Spaces struct {
Spaces map[string]*Space
etag string
}
// Space holds the received shares of one user in one space
type Space struct {
States map[string]*State
}
// State holds the state information of a received share
type State struct {
State collaboration.ShareState
MountPoint *provider.Reference
Hidden bool
}
// New returns a new Cache instance
func New(s metadata.Storage, ttl time.Duration) Cache {
return Cache{
ReceivedSpaces: mtimesyncedcache.Map[string, *Spaces]{},
storage: s,
ttl: ttl,
lockMap: sync.Map{},
}
}
func (c *Cache) lockUser(userID string) func() {
v, _ := c.lockMap.LoadOrStore(userID, &sync.Mutex{})
lock := v.(*sync.Mutex)
lock.Lock()
return func() { lock.Unlock() }
}
// Add adds a new entry to the cache
func (c *Cache) Add(ctx context.Context, userID, spaceID string, rs *collaboration.ReceivedShare) error {
ctx, span := appctx.GetTracerProvider(ctx).Tracer(tracerName).Start(ctx, "Grab lock")
unlock := c.lockUser(userID)
span.End()
span.SetAttributes(attribute.String("cs3.userid", userID))
defer unlock()
if _, ok := c.ReceivedSpaces.Load(userID); !ok {
err := c.syncWithLock(ctx, userID)
if err != nil {
return err
}
}
ctx, span = appctx.GetTracerProvider(ctx).Tracer(tracerName).Start(ctx, "Add")
defer span.End()
span.SetAttributes(attribute.String("cs3.userid", userID), attribute.String("cs3.spaceid", spaceID))
persistFunc := func() error {
c.initializeIfNeeded(userID, spaceID)
rss, _ := c.ReceivedSpaces.Load(userID)
receivedSpace := rss.Spaces[spaceID]
if receivedSpace.States == nil {
receivedSpace.States = map[string]*State{}
}
receivedSpace.States[rs.Share.Id.GetOpaqueId()] = &State{
State: rs.State,
MountPoint: rs.MountPoint,
Hidden: rs.Hidden,
}
return c.persist(ctx, userID)
}
log := appctx.GetLogger(ctx).With().
Str("hostname", os.Getenv("HOSTNAME")).
Str("userID", userID).
Str("spaceID", spaceID).Logger()
var err error
for retries := 100; retries > 0; retries-- {
err = persistFunc()
switch err.(type) {
case nil:
span.SetStatus(codes.Ok, "")
return nil
case errtypes.Aborted:
log.Debug().Msg("aborted when persisting added received share: etag changed. retrying...")
// this is the expected status code from the server when the if-match etag check fails
// continue with sync below
case errtypes.PreconditionFailed:
log.Debug().Msg("precondition failed when persisting added received share: etag changed. retrying...")
// actually, this is the wrong status code and we treat it like errtypes.Aborted because of inconsistencies on the server side
// continue with sync below
case errtypes.AlreadyExists:
log.Debug().Msg("already exists when persisting added received share. retrying...")
// CS3 uses an already exists error instead of precondition failed when using an If-None-Match=* header / IfExists flag in the InitiateFileUpload call.
// Thas happens when the cache thinks there is no file.
// continue with sync below
default:
span.SetStatus(codes.Error, fmt.Sprintf("persisting added received share failed. giving up: %s", err.Error()))
log.Error().Err(err).Msg("persisting added received share failed")
return err
}
if err := c.syncWithLock(ctx, userID); err != nil {
span.RecordError(err)
span.SetStatus(codes.Error, err.Error())
log.Error().Err(err).Msg("persisting added received share failed. giving up.")
return err
}
}
return err
}
// Get returns one entry from the cache
func (c *Cache) Get(ctx context.Context, userID, spaceID, shareID string) (*State, error) {
ctx, span := appctx.GetTracerProvider(ctx).Tracer(tracerName).Start(ctx, "Grab lock")
unlock := c.lockUser(userID)
span.End()
span.SetAttributes(attribute.String("cs3.userid", userID))
defer unlock()
err := c.syncWithLock(ctx, userID)
if err != nil {
return nil, err
}
rss, ok := c.ReceivedSpaces.Load(userID)
if !ok || rss.Spaces[spaceID] == nil {
return nil, nil
}
return rss.Spaces[spaceID].States[shareID], nil
}
// Remove removes an entry from the cache
func (c *Cache) Remove(ctx context.Context, userID, spaceID, shareID string) error {
ctx, span := appctx.GetTracerProvider(ctx).Tracer(tracerName).Start(ctx, "Grab lock")
unlock := c.lockUser(userID)
span.End()
span.SetAttributes(attribute.String("cs3.userid", userID))
defer unlock()
ctx, span = appctx.GetTracerProvider(ctx).Tracer(tracerName).Start(ctx, "Add")
defer span.End()
span.SetAttributes(attribute.String("cs3.userid", userID), attribute.String("cs3.spaceid", spaceID))
persistFunc := func() error {
c.initializeIfNeeded(userID, spaceID)
rss, _ := c.ReceivedSpaces.Load(userID)
receivedSpace := rss.Spaces[spaceID]
if receivedSpace.States == nil {
receivedSpace.States = map[string]*State{}
}
delete(receivedSpace.States, shareID)
if len(receivedSpace.States) == 0 {
delete(rss.Spaces, spaceID)
}
return c.persist(ctx, userID)
}
log := appctx.GetLogger(ctx).With().
Str("hostname", os.Getenv("HOSTNAME")).
Str("userID", userID).
Str("spaceID", spaceID).Logger()
var err error
for retries := 100; retries > 0; retries-- {
err = persistFunc()
switch err.(type) {
case nil:
span.SetStatus(codes.Ok, "")
return nil
case errtypes.Aborted:
log.Debug().Msg("aborted when persisting added received share: etag changed. retrying...")
// this is the expected status code from the server when the if-match etag check fails
// continue with sync below
case errtypes.PreconditionFailed:
log.Debug().Msg("precondition failed when persisting added received share: etag changed. retrying...")
// actually, this is the wrong status code and we treat it like errtypes.Aborted because of inconsistencies on the server side
// continue with sync below
case errtypes.AlreadyExists:
log.Debug().Msg("already exists when persisting added received share. retrying...")
// CS3 uses an already exists error instead of precondition failed when using an If-None-Match=* header / IfExists flag in the InitiateFileUpload call.
// Thas happens when the cache thinks there is no file.
// continue with sync below
default:
span.SetStatus(codes.Error, fmt.Sprintf("persisting added received share failed. giving up: %s", err.Error()))
log.Error().Err(err).Msg("persisting added received share failed")
return err
}
if err := c.syncWithLock(ctx, userID); err != nil {
span.RecordError(err)
span.SetStatus(codes.Error, err.Error())
log.Error().Err(err).Msg("persisting added received share failed. giving up.")
return err
}
}
return err
}
// List returns a list of received shares for a given user
// The return list is guaranteed to be thread-safe
func (c *Cache) List(ctx context.Context, userID string) (map[string]*Space, error) {
ctx, span := appctx.GetTracerProvider(ctx).Tracer(tracerName).Start(ctx, "Grab lock")
unlock := c.lockUser(userID)
span.End()
span.SetAttributes(attribute.String("cs3.userid", userID))
defer unlock()
err := c.syncWithLock(ctx, userID)
if err != nil {
return nil, err
}
spaces := map[string]*Space{}
rss, _ := c.ReceivedSpaces.Load(userID)
for spaceID, space := range rss.Spaces {
spaceCopy := &Space{
States: map[string]*State{},
}
for shareID, state := range space.States {
spaceCopy.States[shareID] = &State{
State: state.State,
MountPoint: state.MountPoint,
Hidden: state.Hidden,
}
}
spaces[spaceID] = spaceCopy
}
return spaces, nil
}
func (c *Cache) syncWithLock(ctx context.Context, userID string) error {
ctx, span := appctx.GetTracerProvider(ctx).Tracer(tracerName).Start(ctx, "Sync")
defer span.End()
span.SetAttributes(attribute.String("cs3.userid", userID))
log := appctx.GetLogger(ctx).With().Str("userID", userID).Logger()
c.initializeIfNeeded(userID, "")
jsonPath := userJSONPath(userID)
span.AddEvent("updating cache")
// - update cached list of created shares for the user in memory if changed
rss, _ := c.ReceivedSpaces.Load(userID)
dlres, err := c.storage.Download(ctx, metadata.DownloadRequest{
Path: jsonPath,
IfNoneMatch: []string{rss.etag},
})
switch err.(type) {
case nil:
span.AddEvent("updating local cache")
case errtypes.NotFound:
span.SetStatus(codes.Ok, "")
return nil
case errtypes.NotModified:
span.SetStatus(codes.Ok, "")
return nil
default:
span.SetStatus(codes.Error, fmt.Sprintf("Failed to download the received share: %s", err.Error()))
log.Error().Err(err).Msg("Failed to download the received share")
return err
}
newSpaces := &Spaces{}
err = json.Unmarshal(dlres.Content, newSpaces)
if err != nil {
span.SetStatus(codes.Error, fmt.Sprintf("Failed to unmarshal the received share: %s", err.Error()))
log.Error().Err(err).Msg("Failed to unmarshal the received share")
return err
}
newSpaces.etag = dlres.Etag
c.ReceivedSpaces.Store(userID, newSpaces)
span.SetStatus(codes.Ok, "")
return nil
}
// persist persists the data for one user to the storage
func (c *Cache) persist(ctx context.Context, userID string) error {
ctx, span := appctx.GetTracerProvider(ctx).Tracer(tracerName).Start(ctx, "Persist")
defer span.End()
span.SetAttributes(attribute.String("cs3.userid", userID))
rss, ok := c.ReceivedSpaces.Load(userID)
if !ok {
span.SetStatus(codes.Ok, "no received shares")
return nil
}
createdBytes, err := json.Marshal(rss)
if err != nil {
span.RecordError(err)
span.SetStatus(codes.Error, err.Error())
return err
}
jsonPath := userJSONPath(userID)
if err := c.storage.MakeDirIfNotExist(ctx, path.Dir(jsonPath)); err != nil {
span.RecordError(err)
span.SetStatus(codes.Error, err.Error())
return err
}
ur := metadata.UploadRequest{
Path: jsonPath,
Content: createdBytes,
IfMatchEtag: rss.etag,
}
// when there is no etag in memory make sure the file has not been created on the server, see https://www.rfc-editor.org/rfc/rfc9110#field.if-match
// > If the field value is "*", the condition is false if the origin server has a current representation for the target resource.
if rss.etag == "" {
ur.IfNoneMatch = []string{"*"}
}
res, err := c.storage.Upload(ctx, ur)
if err != nil {
span.RecordError(err)
span.SetStatus(codes.Error, err.Error())
return err
}
rss.etag = res.Etag
span.SetStatus(codes.Ok, "")
return nil
}
func userJSONPath(userID string) string {
return filepath.Join("/users", userID, "received.json")
}
func (c *Cache) initializeIfNeeded(userID, spaceID string) {
rss, _ := c.ReceivedSpaces.LoadOrStore(userID, &Spaces{Spaces: map[string]*Space{}})
if spaceID != "" && rss.Spaces[spaceID] == nil {
rss.Spaces[spaceID] = &Space{}
c.ReceivedSpaces.Store(userID, rss)
}
}
@@ -0,0 +1,371 @@
// Copyright 2018-2021 CERN
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
//
// In applying this license, CERN does not waive the privileges and immunities
// granted to it by virtue of its status as an Intergovernmental Organization
// or submit itself to any jurisdiction.
package sharecache
import (
"context"
"encoding/json"
"fmt"
"os"
"path"
"path/filepath"
"sync"
"time"
"go.opentelemetry.io/otel/attribute"
"go.opentelemetry.io/otel/codes"
"golang.org/x/exp/maps"
"github.com/opencloud-eu/reva/v2/pkg/appctx"
"github.com/opencloud-eu/reva/v2/pkg/errtypes"
"github.com/opencloud-eu/reva/v2/pkg/share/manager/jsoncs3/shareid"
"github.com/opencloud-eu/reva/v2/pkg/storage/utils/decomposedfs/mtimesyncedcache"
"github.com/opencloud-eu/reva/v2/pkg/storage/utils/metadata"
)
// name is the Tracer name used to identify this instrumentation library.
const tracerName = "sharecache"
// Cache caches the list of share ids for users/groups
// It functions as an in-memory cache with a persistence layer
// The storage is sharded by user/group
type Cache struct {
lockMap sync.Map
UserShares mtimesyncedcache.Map[string, *UserShareCache]
storage metadata.Storage
namespace string
filename string
ttl time.Duration
}
// UserShareCache holds the space/share map for one user
type UserShareCache struct {
UserShares map[string]*SpaceShareIDs
Etag string
}
// SpaceShareIDs holds the unique list of share ids for a space
type SpaceShareIDs struct {
IDs map[string]struct{}
}
func (c *Cache) lockUser(userID string) func() {
v, _ := c.lockMap.LoadOrStore(userID, &sync.Mutex{})
lock := v.(*sync.Mutex)
lock.Lock()
return func() { lock.Unlock() }
}
// New returns a new Cache instance
func New(s metadata.Storage, namespace, filename string, ttl time.Duration) Cache {
return Cache{
UserShares: mtimesyncedcache.Map[string, *UserShareCache]{},
storage: s,
namespace: namespace,
filename: filename,
ttl: ttl,
lockMap: sync.Map{},
}
}
// Add adds a share to the cache
func (c *Cache) Add(ctx context.Context, userid, shareID string) error {
ctx, span := appctx.GetTracerProvider(ctx).Tracer(tracerName).Start(ctx, "Grab lock")
unlock := c.lockUser(userid)
span.End()
span.SetAttributes(attribute.String("cs3.userid", userid))
defer unlock()
if _, ok := c.UserShares.Load(userid); !ok {
err := c.syncWithLock(ctx, userid)
if err != nil {
return err
}
}
ctx, span = appctx.GetTracerProvider(ctx).Tracer(tracerName).Start(ctx, "Add")
defer span.End()
span.SetAttributes(attribute.String("cs3.userid", userid), attribute.String("cs3.shareid", shareID))
storageid, spaceid, _ := shareid.Decode(shareID)
ssid := storageid + shareid.IDDelimiter + spaceid
persistFunc := func() error {
c.initializeIfNeeded(userid, ssid)
// add share id
us, _ := c.UserShares.Load(userid)
us.UserShares[ssid].IDs[shareID] = struct{}{}
return c.Persist(ctx, userid)
}
log := appctx.GetLogger(ctx).With().
Str("hostname", os.Getenv("HOSTNAME")).
Str("userID", userid).
Str("shareID", shareID).Logger()
var err error
for retries := 100; retries > 0; retries-- {
err = persistFunc()
switch err.(type) {
case nil:
span.SetStatus(codes.Ok, "")
return nil
case errtypes.Aborted:
log.Debug().Msg("aborted when persisting added share: etag changed. retrying...")
// this is the expected status code from the server when the if-match etag check fails
// continue with sync below
case errtypes.PreconditionFailed:
log.Debug().Msg("precondition failed when persisting added share: etag changed. retrying...")
// actually, this is the wrong status code and we treat it like errtypes.Aborted because of inconsistencies on the server side
// continue with sync below
case errtypes.AlreadyExists:
log.Debug().Msg("already exists when persisting added share. retrying...")
// CS3 uses an already exists error instead of precondition failed when using an If-None-Match=* header / IfExists flag in the InitiateFileUpload call.
// Thas happens when the cache thinks there is no file.
// continue with sync below
default:
span.SetStatus(codes.Error, fmt.Sprintf("persisting added share failed. giving up: %s", err.Error()))
log.Error().Err(err).Msg("persisting added share failed")
return err
}
if err := c.syncWithLock(ctx, userid); err != nil {
span.RecordError(err)
span.SetStatus(codes.Error, err.Error())
log.Error().Err(err).Msg("persisting added share failed. giving up.")
return err
}
}
return err
}
// Remove removes a share for the given user
func (c *Cache) Remove(ctx context.Context, userid, shareID string) error {
ctx, span := appctx.GetTracerProvider(ctx).Tracer(tracerName).Start(ctx, "Grab lock")
unlock := c.lockUser(userid)
span.End()
span.SetAttributes(attribute.String("cs3.userid", userid))
defer unlock()
if _, ok := c.UserShares.Load(userid); ok {
err := c.syncWithLock(ctx, userid)
if err != nil {
return err
}
}
ctx, span = appctx.GetTracerProvider(ctx).Tracer(tracerName).Start(ctx, "Remove")
defer span.End()
span.SetAttributes(attribute.String("cs3.userid", userid), attribute.String("cs3.shareid", shareID))
storageid, spaceid, _ := shareid.Decode(shareID)
ssid := storageid + shareid.IDDelimiter + spaceid
persistFunc := func() error {
us, loaded := c.UserShares.LoadOrStore(userid, &UserShareCache{
UserShares: map[string]*SpaceShareIDs{},
})
if loaded {
// remove share id
delete(us.UserShares[ssid].IDs, shareID)
}
return c.Persist(ctx, userid)
}
log := appctx.GetLogger(ctx).With().
Str("hostname", os.Getenv("HOSTNAME")).
Str("userID", userid).
Str("shareID", shareID).Logger()
var err error
for retries := 100; retries > 0; retries-- {
err = persistFunc()
switch err.(type) {
case nil:
span.SetStatus(codes.Ok, "")
return nil
case errtypes.Aborted:
log.Debug().Msg("aborted when persisting removed share: etag changed. retrying...")
// this is the expected status code from the server when the if-match etag check fails
// continue with sync below
case errtypes.PreconditionFailed:
log.Debug().Msg("precondition failed when persisting removed share: etag changed. retrying...")
// actually, this is the wrong status code and we treat it like errtypes.Aborted because of inconsistencies on the server side
// continue with sync below
case errtypes.AlreadyExists:
log.Debug().Msg("file already existed when persisting removed share. retrying...")
// CS3 uses an already exists error instead of precondition failed when using an If-None-Match=* header / IfExists flag in the InitiateFileUpload call.
// Thas happens when the cache thinks there is no file.
// continue with sync below
default:
span.SetStatus(codes.Error, fmt.Sprintf("persisting removed share failed. giving up: %s", err.Error()))
log.Error().Err(err).Msg("persisting removed share failed")
return err
}
if err := c.syncWithLock(ctx, userid); err != nil {
span.RecordError(err)
span.SetStatus(codes.Error, err.Error())
return err
}
}
return err
}
// List return the list of spaces/shares for the given user/group
func (c *Cache) List(ctx context.Context, userid string) (map[string]SpaceShareIDs, error) {
ctx, span := appctx.GetTracerProvider(ctx).Tracer(tracerName).Start(ctx, "Grab lock")
unlock := c.lockUser(userid)
span.End()
span.SetAttributes(attribute.String("cs3.userid", userid))
defer unlock()
if err := c.syncWithLock(ctx, userid); err != nil {
return nil, err
}
r := map[string]SpaceShareIDs{}
us, ok := c.UserShares.Load(userid)
if !ok {
return r, nil
}
for ssid, cached := range us.UserShares {
r[ssid] = SpaceShareIDs{
IDs: maps.Clone(cached.IDs),
}
}
return r, nil
}
func (c *Cache) syncWithLock(ctx context.Context, userID string) error {
ctx, span := appctx.GetTracerProvider(ctx).Tracer(tracerName).Start(ctx, "Sync")
defer span.End()
span.SetAttributes(attribute.String("cs3.userid", userID))
log := appctx.GetLogger(ctx).With().Str("userID", userID).Logger()
c.initializeIfNeeded(userID, "")
userCreatedPath := c.userCreatedPath(userID)
span.AddEvent("updating cache")
// - update cached list of created shares for the user in memory if changed
dlreq := metadata.DownloadRequest{
Path: userCreatedPath,
}
if us, ok := c.UserShares.Load(userID); ok && us.Etag != "" {
dlreq.IfNoneMatch = []string{us.Etag}
}
dlres, err := c.storage.Download(ctx, dlreq)
switch err.(type) {
case nil:
span.AddEvent("updating local cache")
case errtypes.NotFound:
span.SetStatus(codes.Ok, "")
return nil
case errtypes.NotModified:
span.SetStatus(codes.Ok, "")
return nil
default:
span.SetStatus(codes.Error, fmt.Sprintf("Failed to download the share cache: %s", err.Error()))
log.Error().Err(err).Msg("Failed to download the share cache")
return err
}
newShareCache := &UserShareCache{}
err = json.Unmarshal(dlres.Content, newShareCache)
if err != nil {
span.SetStatus(codes.Error, fmt.Sprintf("Failed to unmarshal the share cache: %s", err.Error()))
log.Error().Err(err).Msg("Failed to unmarshal the share cache")
return err
}
newShareCache.Etag = dlres.Etag
c.UserShares.Store(userID, newShareCache)
span.SetStatus(codes.Ok, "")
return nil
}
// Persist persists the data for one user/group to the storage
func (c *Cache) Persist(ctx context.Context, userid string) error {
ctx, span := appctx.GetTracerProvider(ctx).Tracer(tracerName).Start(ctx, "Persist")
defer span.End()
span.SetAttributes(attribute.String("cs3.userid", userid))
us, ok := c.UserShares.Load(userid)
if !ok {
span.SetStatus(codes.Ok, "no user shares")
return nil
}
createdBytes, err := json.Marshal(us)
if err != nil {
span.RecordError(err)
span.SetStatus(codes.Error, err.Error())
return err
}
jsonPath := c.userCreatedPath(userid)
if err := c.storage.MakeDirIfNotExist(ctx, path.Dir(jsonPath)); err != nil {
span.RecordError(err)
span.SetStatus(codes.Error, err.Error())
return err
}
ur := metadata.UploadRequest{
Path: jsonPath,
Content: createdBytes,
IfMatchEtag: us.Etag,
}
// when there is no etag in memory make sure the file has not been created on the server, see https://www.rfc-editor.org/rfc/rfc9110#field.if-match
// > If the field value is "*", the condition is false if the origin server has a current representation for the target resource.
if us.Etag == "" {
ur.IfNoneMatch = []string{"*"}
}
res, err := c.storage.Upload(ctx, ur)
if err != nil {
span.RecordError(err)
span.SetStatus(codes.Error, err.Error())
return err
}
us.Etag = res.Etag
span.SetStatus(codes.Ok, "")
return nil
}
func (c *Cache) userCreatedPath(userid string) string {
return filepath.Join("/", c.namespace, userid, c.filename)
}
func (c *Cache) initializeIfNeeded(userid, ssid string) {
us, _ := c.UserShares.LoadOrStore(userid, &UserShareCache{
UserShares: map[string]*SpaceShareIDs{},
})
if ssid != "" && us.UserShares[ssid] == nil {
us.UserShares[ssid] = &SpaceShareIDs{
IDs: map[string]struct{}{},
}
}
}
@@ -0,0 +1,45 @@
// Copyright 2018-2021 CERN
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
//
// In applying this license, CERN does not waive the privileges and immunities
// granted to it by virtue of its status as an Intergovernmental Organization
// or submit itself to any jurisdiction.
package shareid
import "strings"
const (
// IDDelimiter is used to separate the providerid, spaceid and shareid
IDDelimiter = ":"
)
// Encode encodes a share id
func Encode(providerID, spaceID, shareID string) string {
return providerID + IDDelimiter + spaceID + IDDelimiter + shareID
}
// Decode decodes an encoded shareid
// share ids are of the format <storageid>:<spaceid>:<shareid>
func Decode(id string) (string, string, string) {
parts := strings.SplitN(id, IDDelimiter, 3)
switch len(parts) {
case 1:
return "", "", parts[0]
case 2:
return parts[0], parts[1], ""
default:
return parts[0], parts[1], parts[2]
}
}
@@ -0,0 +1,29 @@
// Copyright 2018-2021 CERN
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
//
// In applying this license, CERN does not waive the privileges and immunities
// granted to it by virtue of its status as an Intergovernmental Organization
// or submit itself to any jurisdiction.
package loader
import (
// Load core share manager drivers.
_ "github.com/opencloud-eu/reva/v2/pkg/share/manager/cs3"
_ "github.com/opencloud-eu/reva/v2/pkg/share/manager/json"
_ "github.com/opencloud-eu/reva/v2/pkg/share/manager/jsoncs3"
_ "github.com/opencloud-eu/reva/v2/pkg/share/manager/memory"
_ "github.com/opencloud-eu/reva/v2/pkg/share/manager/owncloudsql"
// Add your own here
)
@@ -0,0 +1,402 @@
// Copyright 2018-2021 CERN
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
//
// In applying this license, CERN does not waive the privileges and immunities
// granted to it by virtue of its status as an Intergovernmental Organization
// or submit itself to any jurisdiction.
package memory
import (
"context"
"errors"
"fmt"
"sync"
"sync/atomic"
"time"
ctxpkg "github.com/opencloud-eu/reva/v2/pkg/ctx"
"github.com/opencloud-eu/reva/v2/pkg/share"
"google.golang.org/genproto/protobuf/field_mask"
userv1beta1 "github.com/cs3org/go-cs3apis/cs3/identity/user/v1beta1"
collaboration "github.com/cs3org/go-cs3apis/cs3/sharing/collaboration/v1beta1"
provider "github.com/cs3org/go-cs3apis/cs3/storage/provider/v1beta1"
typespb "github.com/cs3org/go-cs3apis/cs3/types/v1beta1"
"github.com/opencloud-eu/reva/v2/pkg/errtypes"
"github.com/opencloud-eu/reva/v2/pkg/share/manager/registry"
"github.com/opencloud-eu/reva/v2/pkg/utils"
)
var counter uint64
func init() {
registry.Register("memory", New)
}
// New returns a new manager.
func New(c map[string]interface{}) (share.Manager, error) {
state := map[string]map[*collaboration.ShareId]collaboration.ShareState{}
mp := map[string]map[*collaboration.ShareId]*provider.Reference{}
return &manager{
shareState: state,
shareMountPoint: mp,
lock: &sync.Mutex{},
}, nil
}
type manager struct {
lock *sync.Mutex
shares []*collaboration.Share
// shareState contains the share state for a user.
// map["alice"]["share-id"]state.
shareState map[string]map[*collaboration.ShareId]collaboration.ShareState
// shareMountPoint contains the mountpoint of a share for a user.
// map["alice"]["share-id"]reference.
shareMountPoint map[string]map[*collaboration.ShareId]*provider.Reference
}
func (m *manager) add(ctx context.Context, s *collaboration.Share) {
m.lock.Lock()
defer m.lock.Unlock()
m.shares = append(m.shares, s)
}
func (m *manager) Share(ctx context.Context, md *provider.ResourceInfo, g *collaboration.ShareGrant) (*collaboration.Share, error) {
id := atomic.AddUint64(&counter, 1)
user := ctxpkg.ContextMustGetUser(ctx)
now := time.Now().UnixNano()
ts := &typespb.Timestamp{
Seconds: uint64(now / 1000000000),
Nanos: uint32(now % 1000000000),
}
if g.Grantee.Type == provider.GranteeType_GRANTEE_TYPE_USER &&
(utils.UserEqual(g.Grantee.GetUserId(), user.Id) || utils.UserEqual(g.Grantee.GetUserId(), md.Owner)) {
return nil, errtypes.BadRequest("memory: owner/creator and grantee are the same")
}
// check if share already exists.
key := &collaboration.ShareKey{
Owner: md.Owner,
ResourceId: md.Id,
Grantee: g.Grantee,
}
_, err := m.getByKey(ctx, key)
// share already exists
if err == nil {
return nil, errtypes.AlreadyExists(key.String())
}
s := &collaboration.Share{
Id: &collaboration.ShareId{
OpaqueId: fmt.Sprintf("%d", id),
},
ResourceId: md.Id,
Permissions: g.Permissions,
Grantee: g.Grantee,
Owner: md.Owner,
Creator: user.Id,
Ctime: ts,
Mtime: ts,
}
m.add(ctx, s)
return s, nil
}
func (m *manager) getByID(ctx context.Context, id *collaboration.ShareId) (*collaboration.Share, error) {
m.lock.Lock()
defer m.lock.Unlock()
for _, s := range m.shares {
if s.GetId().OpaqueId == id.OpaqueId {
return s, nil
}
}
return nil, errtypes.NotFound(id.String())
}
func (m *manager) getByKey(ctx context.Context, key *collaboration.ShareKey) (*collaboration.Share, error) {
m.lock.Lock()
defer m.lock.Unlock()
for _, s := range m.shares {
if (utils.UserEqual(key.Owner, s.Owner) || utils.UserEqual(key.Owner, s.Creator)) &&
utils.ResourceIDEqual(key.ResourceId, s.ResourceId) && utils.GranteeEqual(key.Grantee, s.Grantee) {
return s, nil
}
}
return nil, errtypes.NotFound(key.String())
}
func (m *manager) get(ctx context.Context, ref *collaboration.ShareReference) (s *collaboration.Share, err error) {
switch {
case ref.GetId() != nil:
s, err = m.getByID(ctx, ref.GetId())
case ref.GetKey() != nil:
s, err = m.getByKey(ctx, ref.GetKey())
default:
err = errtypes.NotFound(ref.String())
}
if err != nil {
return nil, err
}
// check if we are the owner
user := ctxpkg.ContextMustGetUser(ctx)
if share.IsCreatedByUser(s, user) {
return s, nil
}
// or the grantee
if s.Grantee.Type == provider.GranteeType_GRANTEE_TYPE_USER && utils.UserEqual(user.Id, s.Grantee.GetUserId()) {
return s, nil
} else if s.Grantee.Type == provider.GranteeType_GRANTEE_TYPE_GROUP {
// check if all user groups match this share; TODO(labkode): filter shares created by us.
for _, g := range user.Groups {
if g == s.Grantee.GetGroupId().OpaqueId {
return s, nil
}
}
}
// we return not found to not disclose information
return nil, errtypes.NotFound(ref.String())
}
func (m *manager) GetShare(ctx context.Context, ref *collaboration.ShareReference) (*collaboration.Share, error) {
share, err := m.get(ctx, ref)
if err != nil {
return nil, err
}
return share, nil
}
func (m *manager) Unshare(ctx context.Context, ref *collaboration.ShareReference) error {
m.lock.Lock()
defer m.lock.Unlock()
user := ctxpkg.ContextMustGetUser(ctx)
for i, s := range m.shares {
if sharesEqual(ref, s) {
if share.IsCreatedByUser(s, user) {
m.shares[len(m.shares)-1], m.shares[i] = m.shares[i], m.shares[len(m.shares)-1]
m.shares = m.shares[:len(m.shares)-1]
return nil
}
}
}
return errtypes.NotFound(ref.String())
}
func sharesEqual(ref *collaboration.ShareReference, s *collaboration.Share) bool {
if ref.GetId() != nil && s.Id != nil {
if ref.GetId().OpaqueId == s.Id.OpaqueId {
return true
}
} else if ref.GetKey() != nil {
if (utils.UserEqual(ref.GetKey().Owner, s.Owner) || utils.UserEqual(ref.GetKey().Owner, s.Creator)) &&
utils.ResourceIDEqual(ref.GetKey().ResourceId, s.ResourceId) && utils.GranteeEqual(ref.GetKey().Grantee, s.Grantee) {
return true
}
}
return false
}
func (m *manager) UpdateShare(ctx context.Context, ref *collaboration.ShareReference, p *collaboration.SharePermissions, updated *collaboration.Share, fieldMask *field_mask.FieldMask) (*collaboration.Share, error) {
m.lock.Lock()
defer m.lock.Unlock()
user := ctxpkg.ContextMustGetUser(ctx)
var shareRef *collaboration.ShareReference
if ref != nil {
shareRef = ref
} else if updated != nil {
shareRef = &collaboration.ShareReference{
Spec: &collaboration.ShareReference_Id{
Id: updated.Id,
},
}
}
for i, s := range m.shares {
if sharesEqual(shareRef, s) {
if share.IsCreatedByUser(s, user) {
now := time.Now().UnixNano()
if p != nil {
m.shares[i].Permissions = p
}
if fieldMask != nil {
for _, path := range fieldMask.Paths {
switch path {
case "permissions":
m.shares[i].Permissions = updated.Permissions
case "expiration":
m.shares[i].Expiration = updated.Expiration
default:
return nil, errtypes.NotSupported("updating " + path + " is not supported")
}
}
}
m.shares[i].Mtime = &typespb.Timestamp{
Seconds: uint64(now / 1000000000),
Nanos: uint32(now % 1000000000),
}
return m.shares[i], nil
}
}
}
return nil, errtypes.NotFound(ref.String())
}
func (m *manager) ListShares(ctx context.Context, filters []*collaboration.Filter) ([]*collaboration.Share, error) {
var ss []*collaboration.Share
m.lock.Lock()
defer m.lock.Unlock()
user := ctxpkg.ContextMustGetUser(ctx)
for _, s := range m.shares {
if share.IsCreatedByUser(s, user) {
// no filter we return earlier
if len(filters) == 0 {
ss = append(ss, s)
continue
}
// check filters
if share.MatchesFilters(s, filters) {
ss = append(ss, s)
}
}
}
return ss, nil
}
// we list the shares that are targeted to the user in context or to the user groups.
func (m *manager) ListReceivedShares(ctx context.Context, filters []*collaboration.Filter, forUser *userv1beta1.UserId) ([]*collaboration.ReceivedShare, error) {
var rss []*collaboration.ReceivedShare
m.lock.Lock()
defer m.lock.Unlock()
user := ctxpkg.ContextMustGetUser(ctx)
if user.GetId().GetType() == userv1beta1.UserType_USER_TYPE_SERVICE {
// TODO: gateway missing!
return nil, errors.New("can't use inmem share manager and service accounts")
}
for _, s := range m.shares {
if share.IsCreatedByUser(s, user) || !share.IsGrantedToUser(s, user) {
// omit shares created by the user or shares the user can't access
continue
}
if len(filters) == 0 {
rs := m.convert(ctx, s)
rss = append(rss, rs)
continue
}
if share.MatchesFilters(s, filters) {
rs := m.convert(ctx, s)
rss = append(rss, rs)
}
}
return rss, nil
}
// convert must be called in a lock-controlled block.
func (m *manager) convert(ctx context.Context, s *collaboration.Share) *collaboration.ReceivedShare {
rs := &collaboration.ReceivedShare{
Share: s,
State: collaboration.ShareState_SHARE_STATE_PENDING,
}
user := ctxpkg.ContextMustGetUser(ctx)
if v, ok := m.shareState[user.Id.String()]; ok {
if state, ok := v[s.Id]; ok {
rs.State = state
}
}
if v, ok := m.shareMountPoint[user.Id.String()]; ok {
if mp, ok := v[s.Id]; ok {
rs.MountPoint = mp
}
}
return rs
}
func (m *manager) GetReceivedShare(ctx context.Context, ref *collaboration.ShareReference) (*collaboration.ReceivedShare, error) {
return m.getReceived(ctx, ref)
}
func (m *manager) getReceived(ctx context.Context, ref *collaboration.ShareReference) (*collaboration.ReceivedShare, error) {
m.lock.Lock()
defer m.lock.Unlock()
user := ctxpkg.ContextMustGetUser(ctx)
for _, s := range m.shares {
if sharesEqual(ref, s) {
if user.GetId().GetType() == userv1beta1.UserType_USER_TYPE_SERVICE || share.IsGrantedToUser(s, user) {
rs := m.convert(ctx, s)
return rs, nil
}
}
}
return nil, errtypes.NotFound(ref.String())
}
func (m *manager) UpdateReceivedShare(ctx context.Context, receivedShare *collaboration.ReceivedShare, fieldMask *field_mask.FieldMask, forUser *userv1beta1.UserId) (*collaboration.ReceivedShare, error) {
rs, err := m.getReceived(ctx, &collaboration.ShareReference{Spec: &collaboration.ShareReference_Id{Id: receivedShare.Share.Id}})
if err != nil {
return nil, err
}
m.lock.Lock()
defer m.lock.Unlock()
for i := range fieldMask.Paths {
switch fieldMask.Paths[i] {
case "state":
rs.State = receivedShare.State
case "mount_point":
rs.MountPoint = receivedShare.MountPoint
case "hidden":
continue
default:
return nil, errtypes.NotSupported("updating " + fieldMask.Paths[i] + " is not supported")
}
}
u := ctxpkg.ContextMustGetUser(ctx)
uid := u.GetId().String()
if u.GetId().GetType() == userv1beta1.UserType_USER_TYPE_SERVICE {
uid = forUser.String()
}
// Persist state
if v, ok := m.shareState[uid]; ok {
v[rs.Share.Id] = rs.State
m.shareState[uid] = v
} else {
a := map[*collaboration.ShareId]collaboration.ShareState{
rs.Share.Id: rs.State,
}
m.shareState[uid] = a
}
// Persist mount point
if v, ok := m.shareMountPoint[uid]; ok {
v[rs.Share.Id] = rs.MountPoint
m.shareMountPoint[uid] = v
} else {
a := map[*collaboration.ShareId]*provider.Reference{
rs.Share.Id: rs.MountPoint,
}
m.shareMountPoint[uid] = a
}
return rs, nil
}
@@ -0,0 +1,301 @@
// Copyright 2018-2021 CERN
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
//
// In applying this license, CERN does not waive the privileges and immunities
// granted to it by virtue of its status as an Intergovernmental Organization
// or submit itself to any jurisdiction.
package owncloudsql
import (
"context"
"strings"
"time"
grouppb "github.com/cs3org/go-cs3apis/cs3/identity/group/v1beta1"
userpb "github.com/cs3org/go-cs3apis/cs3/identity/user/v1beta1"
userprovider "github.com/cs3org/go-cs3apis/cs3/identity/user/v1beta1"
rpc "github.com/cs3org/go-cs3apis/cs3/rpc/v1beta1"
collaboration "github.com/cs3org/go-cs3apis/cs3/sharing/collaboration/v1beta1"
provider "github.com/cs3org/go-cs3apis/cs3/storage/provider/v1beta1"
typespb "github.com/cs3org/go-cs3apis/cs3/types/v1beta1"
"github.com/jellydator/ttlcache/v2"
"github.com/opencloud-eu/reva/v2/pkg/conversions"
"github.com/opencloud-eu/reva/v2/pkg/rgrpc/status"
"github.com/opencloud-eu/reva/v2/pkg/rgrpc/todo/pool"
"github.com/opencloud-eu/reva/v2/pkg/utils"
)
// DBShare stores information about user and public shares.
type DBShare struct {
ID string
UIDOwner string
UIDInitiator string
ItemStorage string
FileSource string
ShareWith string
Token string
Expiration string
Permissions int
ShareType int
ShareName string
STime int
FileTarget string
RejectedBy string
State int
Parent int
}
// UserConverter describes an interface for converting user ids to names and back
type UserConverter interface {
UserNameToUserID(ctx context.Context, username string) (*userpb.UserId, error)
UserIDToUserName(ctx context.Context, userid *userpb.UserId) (string, error)
GetUser(userid *userpb.UserId) (*userpb.User, error)
}
// GatewayUserConverter converts usernames and ids using the gateway
type GatewayUserConverter struct {
gwAddr string
IDCache *ttlcache.Cache
NameCache *ttlcache.Cache
}
// NewGatewayUserConverter returns a instance of GatewayUserConverter
func NewGatewayUserConverter(gwAddr string) *GatewayUserConverter {
IDCache := ttlcache.NewCache()
_ = IDCache.SetTTL(30 * time.Second)
IDCache.SkipTTLExtensionOnHit(true)
NameCache := ttlcache.NewCache()
_ = NameCache.SetTTL(30 * time.Second)
NameCache.SkipTTLExtensionOnHit(true)
return &GatewayUserConverter{
gwAddr: gwAddr,
IDCache: IDCache,
NameCache: NameCache,
}
}
// UserIDToUserName converts a user ID to an username
func (c *GatewayUserConverter) UserIDToUserName(ctx context.Context, userid *userpb.UserId) (string, error) {
username, err := c.NameCache.Get(userid.String())
if err == nil {
return username.(string), nil
}
gwConn, err := pool.GetGatewayServiceClient(c.gwAddr)
if err != nil {
return "", err
}
getUserResponse, err := gwConn.GetUser(ctx, &userprovider.GetUserRequest{
UserId: userid,
SkipFetchingUserGroups: true,
})
if err != nil {
return "", err
}
if getUserResponse.Status.Code != rpc.Code_CODE_OK {
return "", status.NewErrorFromCode(getUserResponse.Status.Code, "gateway")
}
_ = c.NameCache.Set(userid.String(), getUserResponse.User.Username)
return getUserResponse.User.Username, nil
}
// UserNameToUserID converts a username to an user ID
func (c *GatewayUserConverter) UserNameToUserID(ctx context.Context, username string) (*userpb.UserId, error) {
id, err := c.IDCache.Get(username)
if err == nil {
return id.(*userpb.UserId), nil
}
gwConn, err := pool.GetGatewayServiceClient(c.gwAddr)
if err != nil {
return nil, err
}
getUserResponse, err := gwConn.GetUserByClaim(ctx, &userpb.GetUserByClaimRequest{
Claim: "username",
Value: username,
SkipFetchingUserGroups: true,
})
if err != nil {
return nil, err
}
if getUserResponse.Status.Code != rpc.Code_CODE_OK {
return nil, status.NewErrorFromCode(getUserResponse.Status.Code, "gateway")
}
_ = c.IDCache.Set(username, getUserResponse.User.Id)
return getUserResponse.User.Id, nil
}
// GetUser gets the user
func (c *GatewayUserConverter) GetUser(userid *userpb.UserId) (*userpb.User, error) {
gwc, err := pool.GetGatewayServiceClient(c.gwAddr)
if err != nil {
return nil, err
}
return utils.GetUser(userid, gwc)
}
func (m *mgr) formatGrantee(ctx context.Context, g *provider.Grantee) (int, string, error) {
var granteeType int
var formattedID string
switch g.Type {
case provider.GranteeType_GRANTEE_TYPE_USER:
granteeType = 0
var err error
formattedID, err = m.userConverter.UserIDToUserName(ctx, g.GetUserId())
if err != nil {
return 0, "", err
}
case provider.GranteeType_GRANTEE_TYPE_GROUP:
granteeType = 1
formattedID = formatGroupID(g.GetGroupId())
default:
granteeType = -1
}
return granteeType, formattedID, nil
}
func (m *mgr) extractGrantee(ctx context.Context, t int, g string) (*provider.Grantee, error) {
var grantee provider.Grantee
switch t {
case 0:
userid, err := m.userConverter.UserNameToUserID(ctx, g)
if err != nil {
return nil, err
}
grantee.Type = provider.GranteeType_GRANTEE_TYPE_USER
grantee.Id = &provider.Grantee_UserId{UserId: userid}
case 1, 2:
grantee.Type = provider.GranteeType_GRANTEE_TYPE_GROUP
grantee.Id = &provider.Grantee_GroupId{GroupId: extractGroupID(g)}
default:
grantee.Type = provider.GranteeType_GRANTEE_TYPE_INVALID
}
return &grantee, nil
}
func resourceTypeToItem(r provider.ResourceType) string {
switch r {
case provider.ResourceType_RESOURCE_TYPE_FILE:
return "file"
case provider.ResourceType_RESOURCE_TYPE_CONTAINER:
return "folder"
case provider.ResourceType_RESOURCE_TYPE_REFERENCE:
return "reference"
case provider.ResourceType_RESOURCE_TYPE_SYMLINK:
return "symlink"
default:
return ""
}
}
func sharePermToInt(p *provider.ResourcePermissions) int {
return int(conversions.RoleFromResourcePermissions(p, false).OCSPermissions())
}
func intTosharePerm(p int) (*provider.ResourcePermissions, error) {
perms, err := conversions.NewPermissions(p)
if err != nil {
return nil, err
}
return conversions.RoleFromOCSPermissions(perms, nil).CS3ResourcePermissions(), nil
}
func intToShareState(g int) collaboration.ShareState {
switch g {
case 0:
return collaboration.ShareState_SHARE_STATE_ACCEPTED
case 1:
return collaboration.ShareState_SHARE_STATE_PENDING
case 2:
return collaboration.ShareState_SHARE_STATE_REJECTED
default:
return collaboration.ShareState_SHARE_STATE_INVALID
}
}
func formatUserID(u *userpb.UserId) string {
return u.OpaqueId
}
func formatGroupID(u *grouppb.GroupId) string {
return u.OpaqueId
}
func extractGroupID(u string) *grouppb.GroupId {
return &grouppb.GroupId{OpaqueId: u}
}
func (m *mgr) convertToCS3Share(ctx context.Context, s DBShare, storageMountID string) (*collaboration.Share, error) {
ts := &typespb.Timestamp{
Seconds: uint64(s.STime),
}
permissions, err := intTosharePerm(s.Permissions)
if err != nil {
return nil, err
}
grantee, err := m.extractGrantee(ctx, s.ShareType, s.ShareWith)
if err != nil {
return nil, err
}
owner, err := m.userConverter.UserNameToUserID(ctx, s.UIDOwner)
if err != nil {
return nil, err
}
var creator *userpb.UserId
if s.UIDOwner == s.UIDInitiator {
creator = owner
} else {
creator, err = m.userConverter.UserNameToUserID(ctx, s.UIDOwner)
if err != nil {
return nil, err
}
}
return &collaboration.Share{
Id: &collaboration.ShareId{
OpaqueId: s.ID,
},
ResourceId: &provider.ResourceId{
SpaceId: s.ItemStorage,
OpaqueId: s.FileSource,
},
Permissions: &collaboration.SharePermissions{Permissions: permissions},
Grantee: grantee,
Owner: owner,
Creator: creator,
Ctime: ts,
Mtime: ts,
}, nil
}
func (m *mgr) convertToCS3ReceivedShare(ctx context.Context, s DBShare, storageMountID string) (*collaboration.ReceivedShare, error) {
share, err := m.convertToCS3Share(ctx, s, storageMountID)
if err != nil {
return nil, err
}
var state collaboration.ShareState
if s.RejectedBy != "" {
state = collaboration.ShareState_SHARE_STATE_REJECTED
} else {
state = intToShareState(s.State)
}
return &collaboration.ReceivedShare{
Share: share,
State: state,
MountPoint: &provider.Reference{Path: strings.TrimLeft(s.FileTarget, "/")},
}, nil
}
@@ -0,0 +1,674 @@
// Copyright 2018-2021 CERN
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
//
// In applying this license, CERN does not waive the privileges and immunities
// granted to it by virtue of its status as an Intergovernmental Organization
// or submit itself to any jurisdiction.
package owncloudsql
import (
"context"
"database/sql"
"fmt"
"path"
"strconv"
"strings"
"time"
userpb "github.com/cs3org/go-cs3apis/cs3/identity/user/v1beta1"
collaboration "github.com/cs3org/go-cs3apis/cs3/sharing/collaboration/v1beta1"
provider "github.com/cs3org/go-cs3apis/cs3/storage/provider/v1beta1"
typespb "github.com/cs3org/go-cs3apis/cs3/types/v1beta1"
"github.com/mitchellh/mapstructure"
ctxpkg "github.com/opencloud-eu/reva/v2/pkg/ctx"
"github.com/opencloud-eu/reva/v2/pkg/errtypes"
"github.com/opencloud-eu/reva/v2/pkg/share"
"github.com/opencloud-eu/reva/v2/pkg/share/manager/registry"
"github.com/opencloud-eu/reva/v2/pkg/utils"
"github.com/pkg/errors"
"google.golang.org/genproto/protobuf/field_mask"
// Provides mysql drivers
_ "github.com/go-sql-driver/mysql"
)
const (
shareTypeUser = 0
shareTypeGroup = 1
)
func init() {
registry.Register("owncloudsql", NewMysql)
}
type config struct {
GatewayAddr string `mapstructure:"gateway_addr"`
StorageMountID string `mapstructure:"storage_mount_id"`
DbUsername string `mapstructure:"db_username"`
DbPassword string `mapstructure:"db_password"`
DbHost string `mapstructure:"db_host"`
DbPort int `mapstructure:"db_port"`
DbName string `mapstructure:"db_name"`
}
type mgr struct {
driver string
db *sql.DB
storageMountID string
userConverter UserConverter
}
// NewMysql returns a new share manager connection to a mysql database
func NewMysql(m map[string]interface{}) (share.Manager, error) {
c, err := parseConfig(m)
if err != nil {
err = errors.Wrap(err, "error creating a new manager")
return nil, err
}
db, err := sql.Open("mysql", fmt.Sprintf("%s:%s@tcp(%s:%d)/%s", c.DbUsername, c.DbPassword, c.DbHost, c.DbPort, c.DbName))
if err != nil {
return nil, err
}
userConverter := NewGatewayUserConverter(c.GatewayAddr)
return New("mysql", db, c.StorageMountID, userConverter)
}
// New returns a new Cache instance connecting to the given sql.DB
func New(driver string, db *sql.DB, storageMountID string, userConverter UserConverter) (share.Manager, error) {
return &mgr{
driver: driver,
db: db,
storageMountID: storageMountID,
userConverter: userConverter,
}, nil
}
func parseConfig(m map[string]interface{}) (*config, error) {
c := &config{}
if err := mapstructure.Decode(m, c); err != nil {
return nil, err
}
return c, nil
}
func (m *mgr) Share(ctx context.Context, md *provider.ResourceInfo, g *collaboration.ShareGrant) (*collaboration.Share, error) {
user := ctxpkg.ContextMustGetUser(ctx)
// do not allow share to myself or the owner if share is for a user
// TODO(labkode): should not this be caught already at the gw level?
if g.Grantee.Type == provider.GranteeType_GRANTEE_TYPE_USER &&
(utils.UserEqual(g.Grantee.GetUserId(), user.Id) || utils.UserEqual(g.Grantee.GetUserId(), md.Owner)) {
return nil, errtypes.BadRequest("owncloudsql: owner/creator and grantee are the same")
}
// check if share already exists.
key := &collaboration.ShareKey{
Owner: md.Owner,
ResourceId: md.Id,
Grantee: g.Grantee,
}
_, err := m.getByKey(ctx, key)
// share already exists
if err == nil {
return nil, errtypes.AlreadyExists(key.String())
}
now := time.Now().Unix()
ts := &typespb.Timestamp{
Seconds: uint64(now),
}
owner, err := m.userConverter.UserIDToUserName(ctx, md.Owner)
if err != nil {
return nil, err
}
shareType, shareWith, err := m.formatGrantee(ctx, g.Grantee)
if err != nil {
return nil, err
}
itemType := resourceTypeToItem(md.Type)
targetPath := path.Join("/", path.Base(md.Path))
permissions := sharePermToInt(g.Permissions.Permissions)
itemSource := md.Id.OpaqueId
fileSource, err := strconv.ParseUint(itemSource, 10, 64)
if err != nil {
// it can be the case that the item source may be a character string
// we leave fileSource blank in that case
fileSource = 0
}
stmtString := "INSERT INTO oc_share (share_type,uid_owner,uid_initiator,item_type,item_source,file_source,permissions,stime,share_with,file_target) VALUES (?,?,?,?,?,?,?,?,?,?)"
stmtValues := []interface{}{shareType, owner, user.Username, itemType, itemSource, fileSource, permissions, now, shareWith, targetPath}
stmt, err := m.db.Prepare(stmtString)
if err != nil {
return nil, err
}
result, err := stmt.ExecContext(ctx, stmtValues...)
if err != nil {
return nil, err
}
lastID, err := result.LastInsertId()
if err != nil {
return nil, err
}
return &collaboration.Share{
Id: &collaboration.ShareId{
OpaqueId: strconv.FormatInt(lastID, 10),
},
ResourceId: md.Id,
Permissions: g.Permissions,
Grantee: g.Grantee,
Owner: md.Owner,
Creator: user.Id,
Ctime: ts,
Mtime: ts,
}, nil
}
func (m *mgr) GetShare(ctx context.Context, ref *collaboration.ShareReference) (*collaboration.Share, error) {
var s *collaboration.Share
var err error
switch {
case ref.GetId() != nil:
s, err = m.getByID(ctx, ref.GetId())
case ref.GetKey() != nil:
s, err = m.getByKey(ctx, ref.GetKey())
default:
err = errtypes.NotFound(ref.String())
}
if err != nil {
return nil, err
}
return s, nil
}
func (m *mgr) Unshare(ctx context.Context, ref *collaboration.ShareReference) error {
uid := ctxpkg.ContextMustGetUser(ctx).Username
var query string
params := []interface{}{}
switch {
case ref.GetId() != nil:
query = "DELETE FROM oc_share where id=? AND (uid_owner=? or uid_initiator=?)"
params = append(params, ref.GetId().OpaqueId, uid, uid)
case ref.GetKey() != nil:
key := ref.GetKey()
shareType, shareWith, err := m.formatGrantee(ctx, key.Grantee)
if err != nil {
return err
}
owner := formatUserID(key.Owner)
query = "DELETE FROM oc_share WHERE uid_owner=? AND file_source=? AND share_type=? AND share_with=? AND (uid_owner=? or uid_initiator=?)"
params = append(params, owner, key.ResourceId.OpaqueId, shareType, shareWith, uid, uid)
default:
return errtypes.NotFound(ref.String())
}
stmt, err := m.db.Prepare(query)
if err != nil {
return err
}
res, err := stmt.ExecContext(ctx, params...)
if err != nil {
return err
}
rowCnt, err := res.RowsAffected()
if err != nil {
return err
}
if rowCnt == 0 {
return errtypes.NotFound(ref.String())
}
return nil
}
func (m *mgr) UpdateShare(ctx context.Context, ref *collaboration.ShareReference, p *collaboration.SharePermissions, updated *collaboration.Share, fieldMask *field_mask.FieldMask) (*collaboration.Share, error) {
permissions := sharePermToInt(p.Permissions)
uid := ctxpkg.ContextMustGetUser(ctx).Username
var query string
params := []interface{}{}
switch {
case ref.GetId() != nil:
query = "update oc_share set permissions=?,stime=? where id=? AND (uid_owner=? or uid_initiator=?)"
params = append(params, permissions, time.Now().Unix(), ref.GetId().OpaqueId, uid, uid)
case ref.GetKey() != nil:
key := ref.GetKey()
shareType, shareWith, err := m.formatGrantee(ctx, key.Grantee)
if err != nil {
return nil, err
}
owner := formatUserID(key.Owner)
query = "update oc_share set permissions=?,stime=? where (uid_owner=? or uid_initiator=?) AND file_source=? AND share_type=? AND share_with=? AND (uid_owner=? or uid_initiator=?)"
params = append(params, permissions, time.Now().Unix(), owner, owner, key.ResourceId.OpaqueId, shareType, shareWith, uid, uid)
default:
return nil, errtypes.NotFound(ref.String())
}
stmt, err := m.db.Prepare(query)
if err != nil {
return nil, err
}
if _, err = stmt.ExecContext(ctx, params...); err != nil {
return nil, err
}
return m.GetShare(ctx, ref)
}
func (m *mgr) ListShares(ctx context.Context, filters []*collaboration.Filter) ([]*collaboration.Share, error) {
uid := ctxpkg.ContextMustGetUser(ctx).Username
query := `
SELECT
coalesce(s.uid_owner, '') as uid_owner, coalesce(s.uid_initiator, '') as uid_initiator,
coalesce(s.share_with, '') as share_with, coalesce(s.file_source, '') as file_source,
s.file_target, s.id, s.stime, s.permissions, s.share_type, fc.storage as storage
FROM oc_share s
LEFT JOIN oc_filecache fc ON fc.fileid = file_source
WHERE (uid_owner=? or uid_initiator=?)
`
params := []interface{}{uid, uid}
var (
filterQuery string
filterParams []interface{}
err error
)
if len(filters) == 0 {
filterQuery += "(share_type=? OR share_type=?)"
params = append(params, shareTypeUser)
params = append(params, shareTypeGroup)
} else {
filterQuery, filterParams, err = translateFilters(filters)
if err != nil {
return nil, err
}
params = append(params, filterParams...)
}
if filterQuery != "" {
query = fmt.Sprintf("%s AND (%s)", query, filterQuery)
}
rows, err := m.db.QueryContext(ctx, query, params...)
if err != nil {
return nil, err
}
defer rows.Close()
var s DBShare
shares := []*collaboration.Share{}
for rows.Next() {
if err := rows.Scan(&s.UIDOwner, &s.UIDInitiator, &s.ShareWith, &s.FileSource, &s.FileTarget, &s.ID, &s.STime, &s.Permissions, &s.ShareType, &s.ItemStorage); err != nil {
continue
}
share, err := m.convertToCS3Share(ctx, s, m.storageMountID)
if err != nil {
return nil, err
}
shares = append(shares, share)
}
if err = rows.Err(); err != nil {
return nil, err
}
return shares, nil
}
// we list the shares that are targeted to the user in context or to the user groups.
func (m *mgr) ListReceivedShares(ctx context.Context, filters []*collaboration.Filter, forUser *userpb.UserId) ([]*collaboration.ReceivedShare, error) {
user := ctxpkg.ContextMustGetUser(ctx)
if user.GetId().GetType() == userpb.UserType_USER_TYPE_SERVICE {
u, err := m.userConverter.GetUser(forUser)
if err != nil {
return nil, errtypes.BadRequest("user not found")
}
user = u
}
uid := user.Username
params := []interface{}{uid, uid, uid}
for _, v := range user.Groups {
params = append(params, v)
}
homeConcat := ""
if m.driver == "mysql" { // mysql concat
homeConcat = "storages.id = CONCAT('home::', s.uid_owner)"
} else { // sqlite3 concat
homeConcat = "storages.id = 'home::' || s.uid_owner"
}
userSelect := ""
if len(user.Groups) > 0 {
userSelect = "AND ((share_type != 1 AND share_with=?) OR (share_type = 1 AND share_with in (?" + strings.Repeat(",?", len(user.Groups)-1) + ")))"
} else {
userSelect = "AND (share_type != 1 AND share_with=?)"
}
query := `
WITH results AS
(
SELECT s.*, storages.numeric_id FROM oc_share s
LEFT JOIN oc_storages storages ON ` + homeConcat + `
WHERE (uid_owner != ? AND uid_initiator != ?) ` + userSelect + `
)
SELECT COALESCE(r.uid_owner, '') AS uid_owner, COALESCE(r.uid_initiator, '') AS uid_initiator, COALESCE(r.share_with, '')
AS share_with, COALESCE(r.file_source, '') AS file_source, COALESCE(r2.file_target, r.file_target), r.id, r.stime, r.permissions, r.share_type, COALESCE(r2.accepted, r.accepted),
r.numeric_id, COALESCE(r.parent, -1) AS parent FROM results r LEFT JOIN results r2 ON r.id = r2.parent WHERE r.parent IS NULL`
filterQuery, filterParams, err := translateFilters(filters)
if err != nil {
return nil, err
}
params = append(params, filterParams...)
if filterQuery != "" {
query = fmt.Sprintf("%s AND (%s)", query, filterQuery)
}
query += ";"
rows, err := m.db.QueryContext(ctx, query, params...)
if err != nil {
return nil, err
}
defer rows.Close()
var s DBShare
shares := []*collaboration.ReceivedShare{}
for rows.Next() {
if err := rows.Scan(&s.UIDOwner, &s.UIDInitiator, &s.ShareWith, &s.FileSource, &s.FileTarget, &s.ID, &s.STime, &s.Permissions, &s.ShareType, &s.State, &s.ItemStorage, &s.Parent); err != nil {
continue
}
share, err := m.convertToCS3ReceivedShare(ctx, s, m.storageMountID)
if err != nil {
return nil, err
}
shares = append(shares, share)
}
if err = rows.Err(); err != nil {
return nil, err
}
return shares, nil
}
func (m *mgr) GetReceivedShare(ctx context.Context, ref *collaboration.ShareReference) (*collaboration.ReceivedShare, error) {
var s *collaboration.ReceivedShare
var err error
switch {
case ref.GetId() != nil:
s, err = m.getReceivedByID(ctx, ref.GetId())
case ref.GetKey() != nil:
s, err = m.getReceivedByKey(ctx, ref.GetKey())
default:
err = errtypes.NotFound(ref.String())
}
if err != nil {
return nil, err
}
return s, nil
}
func (m *mgr) UpdateReceivedShare(ctx context.Context, receivedShare *collaboration.ReceivedShare, fieldMask *field_mask.FieldMask, _ *userpb.UserId) (*collaboration.ReceivedShare, error) {
// TODO: How to inject the uid when a UserId is set? override it in the ctx? Add parameter to GetReceivedShare?
rs, err := m.GetReceivedShare(ctx, &collaboration.ShareReference{Spec: &collaboration.ShareReference_Id{Id: receivedShare.Share.Id}})
if err != nil {
return nil, err
}
fields := []string{}
params := []interface{}{}
for i := range fieldMask.Paths {
switch fieldMask.Paths[i] {
case "state":
rs.State = receivedShare.State
fields = append(fields, "accepted=?")
switch rs.State {
case collaboration.ShareState_SHARE_STATE_REJECTED:
params = append(params, 2)
case collaboration.ShareState_SHARE_STATE_ACCEPTED:
params = append(params, 0)
}
case "mount_point":
fields = append(fields, "file_target=?")
rs.MountPoint = receivedShare.MountPoint
params = append(params, rs.MountPoint.Path)
case "hidden":
continue
default:
return nil, errtypes.NotSupported("updating " + fieldMask.Paths[i] + " is not supported")
}
}
if len(fields) == 0 {
return nil, fmt.Errorf("no valid field provided in the fieldmask")
}
updateReceivedShare := func(column string) error {
query := "update oc_share set "
query += strings.Join(fields, ",")
query += fmt.Sprintf(" where %s=?", column)
params := append(params, rs.Share.Id.OpaqueId)
stmt, err := m.db.Prepare(query)
if err != nil {
return err
}
res, err := stmt.ExecContext(ctx, params...)
if err != nil {
return err
}
affected, err := res.RowsAffected()
if err != nil {
return err
}
if affected < 1 {
return fmt.Errorf("No rows updated")
}
return nil
}
err = updateReceivedShare("parent") // Try to update the child state in case of group shares first
if err != nil {
err = updateReceivedShare("id")
}
if err != nil {
return nil, err
}
return rs, nil
}
func (m *mgr) getByID(ctx context.Context, id *collaboration.ShareId) (*collaboration.Share, error) {
uid := ctxpkg.ContextMustGetUser(ctx).Username
s := DBShare{ID: id.OpaqueId}
query := "select coalesce(uid_owner, '') as uid_owner, coalesce(uid_initiator, '') as uid_initiator, coalesce(share_with, '') as share_with, coalesce(file_source, '') as file_source, file_target, stime, permissions, share_type FROM oc_share WHERE id=? AND (uid_owner=? or uid_initiator=?)"
if err := m.db.QueryRowContext(ctx, query, id.OpaqueId, uid, uid).Scan(&s.UIDOwner, &s.UIDInitiator, &s.ShareWith, &s.FileSource, &s.FileTarget, &s.STime, &s.Permissions, &s.ShareType); err != nil {
if err == sql.ErrNoRows {
return nil, errtypes.NotFound(id.OpaqueId)
}
return nil, err
}
return m.convertToCS3Share(ctx, s, m.storageMountID)
}
func (m *mgr) getByKey(ctx context.Context, key *collaboration.ShareKey) (*collaboration.Share, error) {
owner, err := m.userConverter.UserIDToUserName(ctx, key.Owner)
if err != nil {
return nil, err
}
uid := ctxpkg.ContextMustGetUser(ctx).Username
s := DBShare{}
shareType, shareWith, err := m.formatGrantee(ctx, key.Grantee)
if err != nil {
return nil, err
}
query := "select coalesce(uid_owner, '') as uid_owner, coalesce(uid_initiator, '') as uid_initiator, coalesce(share_with, '') as share_with, coalesce(file_source, '') as file_source, file_target, id, stime, permissions, share_type FROM oc_share WHERE uid_owner=? AND file_source=? AND share_type=? AND share_with=? AND (uid_owner=? or uid_initiator=?)"
if err = m.db.QueryRowContext(ctx, query, owner, key.ResourceId.StorageId, shareType, shareWith, uid, uid).Scan(&s.UIDOwner, &s.UIDInitiator, &s.ShareWith, &s.FileSource, &s.FileTarget, &s.ID, &s.STime, &s.Permissions, &s.ShareType); err != nil {
if err == sql.ErrNoRows {
return nil, errtypes.NotFound(key.String())
}
return nil, err
}
return m.convertToCS3Share(ctx, s, m.storageMountID)
}
func (m *mgr) getReceivedByID(ctx context.Context, id *collaboration.ShareId) (*collaboration.ReceivedShare, error) {
user := ctxpkg.ContextMustGetUser(ctx)
uid := user.Username
params := []interface{}{id.OpaqueId, id.OpaqueId, uid}
for _, v := range user.Groups {
params = append(params, v)
}
homeConcat := ""
if m.driver == "mysql" { // mysql concat
homeConcat = "storages.id = CONCAT('home::', s.uid_owner)"
} else { // sqlite3 concat
homeConcat = "storages.id = 'home::' || s.uid_owner"
}
userSelect := ""
if len(user.Groups) > 0 {
userSelect = "AND ((share_type != 1 AND share_with=?) OR (share_type = 1 AND share_with in (?" + strings.Repeat(",?", len(user.Groups)-1) + ")))"
} else {
userSelect = "AND (share_type != 1 AND share_with=?)"
}
query := `
WITH results AS
(
SELECT s.*, storages.numeric_id
FROM oc_share s
LEFT JOIN oc_storages storages ON ` + homeConcat + `
WHERE s.id=? OR s.parent=? ` + userSelect + `
)
SELECT COALESCE(r.uid_owner, '') AS uid_owner, COALESCE(r.uid_initiator, '') AS uid_initiator, COALESCE(r.share_with, '')
AS share_with, COALESCE(r.file_source, '') AS file_source, COALESCE(r2.file_target, r.file_target), r.id, r.stime, r.permissions, r.share_type, COALESCE(r2.accepted, r.accepted),
r.numeric_id, COALESCE(r.parent, -1) AS parent
FROM results r
LEFT JOIN results r2 ON r.id = r2.parent
WHERE r.parent IS NULL;
`
s := DBShare{}
if err := m.db.QueryRowContext(ctx, query, params...).Scan(&s.UIDOwner, &s.UIDInitiator, &s.ShareWith, &s.FileSource, &s.FileTarget, &s.ID, &s.STime, &s.Permissions, &s.ShareType, &s.State, &s.ItemStorage, &s.Parent); err != nil {
if err == sql.ErrNoRows {
return nil, errtypes.NotFound(id.OpaqueId)
}
return nil, err
}
return m.convertToCS3ReceivedShare(ctx, s, m.storageMountID)
}
func (m *mgr) getReceivedByKey(ctx context.Context, key *collaboration.ShareKey) (*collaboration.ReceivedShare, error) {
user := ctxpkg.ContextMustGetUser(ctx)
uid := user.Username
shareType, shareWith, err := m.formatGrantee(ctx, key.Grantee)
if err != nil {
return nil, err
}
params := []interface{}{uid, formatUserID(key.Owner), key.ResourceId.StorageId, key.ResourceId.OpaqueId, shareType, shareWith, shareWith}
for _, v := range user.Groups {
params = append(params, v)
}
s := DBShare{}
query := "select coalesce(uid_owner, '') as uid_owner, coalesce(uid_initiator, '') as uid_initiator, coalesce(share_with, '') as share_with, coalesce(file_source, '') as file_source, file_target, ts.id, stime, permissions, share_type, accepted FROM oc_share ts WHERE uid_owner=? AND file_source=? AND share_type=? AND share_with=? "
if len(user.Groups) > 0 {
query += "AND (share_with=? OR share_with in (?" + strings.Repeat(",?", len(user.Groups)-1) + "))"
} else {
query += "AND (share_with=?)"
}
if err := m.db.QueryRowContext(ctx, query, params...).Scan(&s.UIDOwner, &s.UIDInitiator, &s.ShareWith, &s.FileSource, &s.FileTarget, &s.ID, &s.STime, &s.Permissions, &s.ShareType, &s.State); err != nil {
if err == sql.ErrNoRows {
return nil, errtypes.NotFound(key.String())
}
return nil, err
}
return m.convertToCS3ReceivedShare(ctx, s, m.storageMountID)
}
func granteeTypeToShareType(granteeType provider.GranteeType) int {
switch granteeType {
case provider.GranteeType_GRANTEE_TYPE_USER:
return shareTypeUser
case provider.GranteeType_GRANTEE_TYPE_GROUP:
return shareTypeGroup
}
return -1
}
// translateFilters translates the filters to sql queries
func translateFilters(filters []*collaboration.Filter) (string, []interface{}, error) {
var (
filterQuery string
params []interface{}
)
groupedFilters := share.GroupFiltersByType(filters)
// If multiple filters of the same type are passed to this function, they need to be combined with the `OR` operator.
// That is why the filters got grouped by type.
// For every given filter type, iterate over the filters and if there are more than one combine them.
// Combine the different filter types using `AND`
var filterCounter = 0
for filterType, filters := range groupedFilters {
switch filterType {
case collaboration.Filter_TYPE_RESOURCE_ID:
filterQuery += "("
for i, f := range filters {
filterQuery += "file_source=?"
params = append(params, f.GetResourceId().OpaqueId)
if i != len(filters)-1 {
filterQuery += " OR "
}
}
filterQuery += ")"
case collaboration.Filter_TYPE_GRANTEE_TYPE:
filterQuery += "("
for i, f := range filters {
filterQuery += "r.share_type=?"
params = append(params, granteeTypeToShareType(f.GetGranteeType()))
if i != len(filters)-1 {
filterQuery += " OR "
}
}
filterQuery += ")"
case collaboration.Filter_TYPE_EXCLUDE_DENIALS:
// TODO this may change once the mapping of permission to share types is completed (cf. pkg/cbox/utils/conversions.go)
filterQuery += "r.permissions > 0"
default:
return "", nil, fmt.Errorf("filter type is not supported")
}
if filterCounter != len(groupedFilters)-1 {
filterQuery += " AND "
}
filterCounter++
}
return filterQuery, params, nil
}
Binary file not shown.
@@ -0,0 +1,34 @@
// Copyright 2018-2021 CERN
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
//
// In applying this license, CERN does not waive the privileges and immunities
// granted to it by virtue of its status as an Intergovernmental Organization
// or submit itself to any jurisdiction.
package registry
import "github.com/opencloud-eu/reva/v2/pkg/share"
// NewFunc is the function that share managers
// should register at init time.
type NewFunc func(map[string]interface{}) (share.Manager, error)
// NewFuncs is a map containing all the registered share managers.
var NewFuncs = map[string]NewFunc{}
// Register registers a new share manager new function.
// Not safe for concurrent use. Safe for use from package init.
func Register(name string, f NewFunc) {
NewFuncs[name] = f
}
+248
View File
@@ -0,0 +1,248 @@
// Copyright 2018-2021 CERN
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
//
// In applying this license, CERN does not waive the privileges and immunities
// granted to it by virtue of its status as an Intergovernmental Organization
// or submit itself to any jurisdiction.
package share
import (
"context"
"time"
userv1beta1 "github.com/cs3org/go-cs3apis/cs3/identity/user/v1beta1"
collaboration "github.com/cs3org/go-cs3apis/cs3/sharing/collaboration/v1beta1"
provider "github.com/cs3org/go-cs3apis/cs3/storage/provider/v1beta1"
types "github.com/cs3org/go-cs3apis/cs3/types/v1beta1"
"github.com/opencloud-eu/reva/v2/pkg/storage/utils/grants"
"github.com/opencloud-eu/reva/v2/pkg/utils"
"google.golang.org/genproto/protobuf/field_mask"
)
const (
// NoState can be used to signal the filter matching functions to ignore the share state.
NoState collaboration.ShareState = -1
)
// Metadata contains Metadata for a share
type Metadata struct {
ETag string
Mtime *types.Timestamp
}
// Manager is the interface that manipulates shares.
type Manager interface {
// Create a new share in fn with the given acl.
Share(ctx context.Context, md *provider.ResourceInfo, g *collaboration.ShareGrant) (*collaboration.Share, error)
// GetShare gets the information for a share by the given ref.
GetShare(ctx context.Context, ref *collaboration.ShareReference) (*collaboration.Share, error)
// Unshare deletes the share pointed by ref.
Unshare(ctx context.Context, ref *collaboration.ShareReference) error
// UpdateShare updates the mode of the given share.
UpdateShare(ctx context.Context, ref *collaboration.ShareReference, p *collaboration.SharePermissions, updated *collaboration.Share, fieldMask *field_mask.FieldMask) (*collaboration.Share, error)
// ListShares returns the shares created by the user. If md is provided is not nil,
// it returns only shares attached to the given resource.
ListShares(ctx context.Context, filters []*collaboration.Filter) ([]*collaboration.Share, error)
// ListReceivedShares returns the list of shares the user has access to. `forUser` parameter for service accounts only
ListReceivedShares(ctx context.Context, filters []*collaboration.Filter, forUser *userv1beta1.UserId) ([]*collaboration.ReceivedShare, error)
// GetReceivedShare returns the information for a received share.
GetReceivedShare(ctx context.Context, ref *collaboration.ShareReference) (*collaboration.ReceivedShare, error)
// UpdateReceivedShare updates the received share with share state.`forUser` parameter for service accounts only
UpdateReceivedShare(ctx context.Context, share *collaboration.ReceivedShare, fieldMask *field_mask.FieldMask, forUser *userv1beta1.UserId) (*collaboration.ReceivedShare, error)
}
// ReceivedShareWithUser holds the relevant information for representing a received share of a user
type ReceivedShareWithUser struct {
UserID *userv1beta1.UserId
ReceivedShare *collaboration.ReceivedShare
}
// DumpableManager defines a share manager which supports dumping its contents
type DumpableManager interface {
Dump(ctx context.Context, shareChan chan<- *collaboration.Share, receivedShareChan chan<- ReceivedShareWithUser) error
}
// LoadableManager defines a share manager which supports loading contents from a dump
type LoadableManager interface {
Load(ctx context.Context, shareChan <-chan *collaboration.Share, receivedShareChan <-chan ReceivedShareWithUser) error
}
// GroupGranteeFilter is an abstraction for creating filter by grantee type group.
func GroupGranteeFilter() *collaboration.Filter {
return &collaboration.Filter{
Type: collaboration.Filter_TYPE_GRANTEE_TYPE,
Term: &collaboration.Filter_GranteeType{
GranteeType: provider.GranteeType_GRANTEE_TYPE_GROUP,
},
}
}
// UserGranteeFilter is an abstraction for creating filter by grantee type user.
func UserGranteeFilter() *collaboration.Filter {
return &collaboration.Filter{
Type: collaboration.Filter_TYPE_GRANTEE_TYPE,
Term: &collaboration.Filter_GranteeType{
GranteeType: provider.GranteeType_GRANTEE_TYPE_USER,
},
}
}
// ResourceIDFilter is an abstraction for creating filter by resource id.
func ResourceIDFilter(id *provider.ResourceId) *collaboration.Filter {
return &collaboration.Filter{
Type: collaboration.Filter_TYPE_RESOURCE_ID,
Term: &collaboration.Filter_ResourceId{
ResourceId: id,
},
}
}
// SpaceIDFilter is an abstraction for creating filter by space id.
func SpaceIDFilter(id string) *collaboration.Filter {
return &collaboration.Filter{
Type: collaboration.Filter_TYPE_SPACE_ID,
Term: &collaboration.Filter_SpaceId{
SpaceId: id,
},
}
}
// StateFilter is an abstraction for creating filter by share state.
func StateFilter(state collaboration.ShareState) *collaboration.Filter {
return &collaboration.Filter{
Type: collaboration.Filter_TYPE_STATE,
Term: &collaboration.Filter_State{
State: state,
},
}
}
// IsCreatedByUser checks if the user is the owner or creator of the share.
func IsCreatedByUser(share *collaboration.Share, user *userv1beta1.User) bool {
return utils.UserEqual(user.Id, share.Owner) || utils.UserEqual(user.Id, share.Creator)
}
// IsGrantedToUser checks if the user is a grantee of the share. Either by a user grant or by a group grant.
func IsGrantedToUser(share *collaboration.Share, user *userv1beta1.User) bool {
if share.Grantee.Type == provider.GranteeType_GRANTEE_TYPE_USER && utils.UserEqual(user.Id, share.Grantee.GetUserId()) {
return true
}
if share.Grantee.Type == provider.GranteeType_GRANTEE_TYPE_GROUP {
// check if any of the user's group is the grantee of the share
for _, g := range user.Groups {
if g == share.Grantee.GetGroupId().OpaqueId {
return true
}
}
}
return false
}
// MatchesFilter tests if the share passes the filter.
func MatchesFilter(share *collaboration.Share, state collaboration.ShareState, filter *collaboration.Filter) bool {
switch filter.Type {
case collaboration.Filter_TYPE_RESOURCE_ID:
return utils.ResourceIDEqual(share.ResourceId, filter.GetResourceId())
case collaboration.Filter_TYPE_GRANTEE_TYPE:
return share.Grantee.Type == filter.GetGranteeType()
case collaboration.Filter_TYPE_EXCLUDE_DENIALS:
// This filter type is used to filter out "denial shares". These are currently implemented by having the permission "0".
// I.e. if the permission is 0 we don't want to show it.
return !grants.PermissionsEqual(share.Permissions.Permissions, &provider.ResourcePermissions{})
case collaboration.Filter_TYPE_SPACE_ID:
return share.ResourceId.SpaceId == filter.GetSpaceId()
case collaboration.Filter_TYPE_STATE:
return state == filter.GetState()
default:
return false
}
}
// MatchesAnyFilter checks if the share passes at least one of the given filters.
func MatchesAnyFilter(share *collaboration.Share, state collaboration.ShareState, filters []*collaboration.Filter) bool {
for _, f := range filters {
if MatchesFilter(share, state, f) {
return true
}
}
return false
}
// MatchesFilters checks if the share passes the given filters.
// Filters of the same type form a disjuntion, a logical OR. Filters of separate type form a conjunction, a logical AND.
// Here is an example:
// (resource_id=1 OR resource_id=2) AND (grantee_type=USER OR grantee_type=GROUP)
func MatchesFilters(share *collaboration.Share, filters []*collaboration.Filter) bool {
if len(filters) == 0 {
return true
}
grouped := GroupFiltersByType(filters)
for _, f := range grouped {
if !MatchesAnyFilter(share, NoState, f) {
return false
}
}
return true
}
// MatchesFiltersWithState checks if the share passes the given filters.
// This can check filter by share state
// Filters of the same type form a disjuntion, a logical OR. Filters of separate type form a conjunction, a logical AND.
// Here is an example:
// (resource_id=1 OR resource_id=2) AND (grantee_type=USER OR grantee_type=GROUP)
func MatchesFiltersWithState(share *collaboration.Share, state collaboration.ShareState, filters []*collaboration.Filter) bool {
if len(filters) == 0 {
return true
}
grouped := GroupFiltersByType(filters)
for _, f := range grouped {
if !MatchesAnyFilter(share, state, f) {
return false
}
}
return true
}
// GroupFiltersByType groups the given filters and returns a map using the filter type as the key.
func GroupFiltersByType(filters []*collaboration.Filter) map[collaboration.Filter_Type][]*collaboration.Filter {
grouped := make(map[collaboration.Filter_Type][]*collaboration.Filter)
for _, f := range filters {
grouped[f.Type] = append(grouped[f.Type], f)
}
return grouped
}
// FilterFiltersByType returns a slice of filters by a given type.
// If no filter with the given type exists within the filters, then an
// empty slice is returned.
func FilterFiltersByType(f []*collaboration.Filter, t collaboration.Filter_Type) []*collaboration.Filter {
return GroupFiltersByType(f)[t]
}
// IsExpired tests whether a share is expired
func IsExpired(s *collaboration.Share) bool {
if e := s.GetExpiration(); e != nil {
expiration := time.Unix(int64(e.Seconds), int64(e.Nanos))
return expiration.Before(time.Now())
}
return false
}