mirror of
https://github.com/opencloud-eu/opencloud.git
synced 2026-09-22 12:08:34 -04:00
chore: bump reva to latest main
This commit is contained in:
1 parent
c2a01e3aad
commit
c198c2e453
30 files changed
+621
-17463
No files matched your search
+14
@@ -53,6 +53,20 @@ func (re *Event) Ack() error {
|
||||
return re.msg.Ack()
|
||||
}
|
||||
|
||||
func (re *Event) Nak() error {
|
||||
if re.msg == nil {
|
||||
return errors.New("cannot nack event without message")
|
||||
}
|
||||
return re.msg.Nak()
|
||||
}
|
||||
|
||||
func (re *Event) Term() error {
|
||||
if re.msg == nil {
|
||||
return errors.New("cannot terminate event without message")
|
||||
}
|
||||
return re.msg.Term()
|
||||
}
|
||||
|
||||
func (re *Event) InProgress() error {
|
||||
if re.msg == nil {
|
||||
return errors.New("cannot mark event as in progress without message")
|
||||
|
||||
Generated
Vendored
+17
@@ -82,6 +82,7 @@ var (
|
||||
events.PostprocessingFinished{},
|
||||
events.PostprocessingStepFinished{},
|
||||
events.RestartPostprocessing{},
|
||||
events.StartPostprocessingStep{},
|
||||
}
|
||||
)
|
||||
|
||||
@@ -441,6 +442,22 @@ func (fs *Decomposedfs) Postprocessing(ch <-chan events.Event) {
|
||||
}); err != nil {
|
||||
sublog.Error().Err(err).Msg("Failed to publish BytesReceived event")
|
||||
}
|
||||
case events.StartPostprocessingStep:
|
||||
sublog := log.With().Str("event", "StartPostprocessingStep").Str("uploadid", ev.UploadID).Logger()
|
||||
if ev.UploadID == "" {
|
||||
sublog.Error().Msg("UploadID is empty, cannot start postprocessing step")
|
||||
continue
|
||||
}
|
||||
session, err := fs.sessionStore.Get(ctx, ev.UploadID)
|
||||
if err != nil {
|
||||
sublog.Error().Err(err).Msg("Failed to get upload")
|
||||
continue
|
||||
}
|
||||
session.SetStatus(upload.SessionStatusProcessing, "started postprocessing step: "+string(ev.StepToStart))
|
||||
err = session.Persist(ctx)
|
||||
if err != nil {
|
||||
sublog.Error().Err(err).Msg("Failed to persist upload session after starting postprocessing step")
|
||||
}
|
||||
case events.PostprocessingStepFinished:
|
||||
sublog := log.With().Str("event", "PostprocessingStepFinished").Str("uploadid", ev.UploadID).Logger()
|
||||
if ev.FinishedStep != events.PPStepAntivirus {
|
||||
|
||||
Generated
Vendored
+1
@@ -102,6 +102,7 @@ const (
|
||||
SpaceImageAttr string = OcPrefix + "space.image"
|
||||
SpaceAliasAttr string = OcPrefix + "space.alias"
|
||||
SpaceTenantIDAttr string = OcPrefix + "space.tenantid"
|
||||
SpaceContentTypeAttr string = OcPrefix + "space.contenttype"
|
||||
|
||||
UserAcePrefix string = "u:"
|
||||
GroupAcePrefix string = "g:"
|
||||
|
||||
+12
-1
@@ -99,6 +99,7 @@ func (fs *Decomposedfs) CreateStorageSpace(ctx context.Context, req *provider.Cr
|
||||
}
|
||||
|
||||
description := utils.ReadPlainFromOpaque(req.Opaque, "description")
|
||||
contentType := utils.ReadPlainFromOpaque(req.Opaque, "contentType")
|
||||
alias := utils.ReadPlainFromOpaque(req.Opaque, "spaceAlias")
|
||||
if alias == "" {
|
||||
alias = templates.WithSpacePropertiesAndUser(u, req.Type, req.Name, spaceID, fs.o.GeneralSpaceAliasTemplate)
|
||||
@@ -192,6 +193,10 @@ func (fs *Decomposedfs) CreateStorageSpace(ctx context.Context, req *provider.Cr
|
||||
metadata.SetString(prefixes.SpaceDescriptionAttr, description)
|
||||
}
|
||||
|
||||
if contentType != "" {
|
||||
metadata.SetString(prefixes.SpaceContentTypeAttr, contentType)
|
||||
}
|
||||
|
||||
if alias != "" {
|
||||
metadata.SetString(prefixes.SpaceAliasAttr, alias)
|
||||
}
|
||||
@@ -607,6 +612,9 @@ func (fs *Decomposedfs) UpdateStorageSpace(ctx context.Context, req *provider.Up
|
||||
if description, ok := space.Opaque.Map["description"]; ok {
|
||||
metadata[prefixes.SpaceDescriptionAttr] = description.Value
|
||||
}
|
||||
if contentType, ok := space.Opaque.Map["contentType"]; ok {
|
||||
metadata[prefixes.SpaceContentTypeAttr] = contentType.Value
|
||||
}
|
||||
if alias := utils.ReadPlainFromOpaque(space.Opaque, "spaceAlias"); alias != "" {
|
||||
metadata.SetString(prefixes.SpaceAliasAttr, alias)
|
||||
}
|
||||
@@ -660,7 +668,7 @@ func (fs *Decomposedfs) UpdateStorageSpace(ctx context.Context, req *provider.Up
|
||||
|
||||
if !permissions.IsManager(sp) {
|
||||
// We are not a space manager. We need to check for additional permissions.
|
||||
k := []string{prefixes.NameAttr, prefixes.SpaceDescriptionAttr}
|
||||
k := []string{prefixes.NameAttr, prefixes.SpaceDescriptionAttr, prefixes.SpaceContentTypeAttr}
|
||||
if !permissions.IsEditor(sp) {
|
||||
k = append(k, prefixes.SpaceReadmeAttr, prefixes.SpaceAliasAttr, prefixes.SpaceImageAttr)
|
||||
}
|
||||
@@ -1118,6 +1126,9 @@ func (fs *Decomposedfs) StorageSpaceFromNode(ctx context.Context, n *node.Node,
|
||||
if sd := spaceAttributes.String(prefixes.SpaceDescriptionAttr); sd != "" {
|
||||
space.Opaque = utils.AppendPlainToOpaque(space.Opaque, "description", sd)
|
||||
}
|
||||
if se := spaceAttributes.String(prefixes.SpaceContentTypeAttr); se != "" {
|
||||
space.Opaque = utils.AppendPlainToOpaque(space.Opaque, "contentType", se)
|
||||
}
|
||||
if sr := spaceAttributes.String(prefixes.SpaceReadmeAttr); sr != "" {
|
||||
space.Opaque = utils.AppendPlainToOpaque(space.Opaque, "readme", storagespace.FormatResourceID(
|
||||
&provider.ResourceId{StorageId: space.Root.StorageId, SpaceId: space.GetRoot().GetSpaceId(), OpaqueId: sr},
|
||||
|
||||
Generated
Vendored
+24
-3
@@ -46,6 +46,12 @@ type DecomposedFsSession struct {
|
||||
info tusd.FileInfo
|
||||
}
|
||||
|
||||
const (
|
||||
SessionStatusUploading = "uploading"
|
||||
SessionStatusProcessing = "processing"
|
||||
SessionStatusFailed = "failed"
|
||||
)
|
||||
|
||||
// Context returns a context with the user, logger and lockid used when initiating the upload session
|
||||
func (session *DecomposedFsSession) Context(ctx context.Context) context.Context { // restore logger from file info
|
||||
sub := session.store.log.With().Int("pid", os.Getpid()).Logger()
|
||||
@@ -307,10 +313,9 @@ func (session *DecomposedFsSession) MTime() time.Time {
|
||||
return t
|
||||
}
|
||||
|
||||
// IsProcessing returns true if all bytes have been received. The session then has entered postprocessing state.
|
||||
// IsProcessing returns true if the upload is in the processing state, meaning that the upload has finished and postprocessing is still running.
|
||||
func (session *DecomposedFsSession) IsProcessing() bool {
|
||||
// We might need a more sophisticated way to determine processing status soon
|
||||
return session.info.Size == session.info.Offset && session.info.MetaData["scanResult"] == ""
|
||||
return session.info.MetaData["status"] == SessionStatusProcessing
|
||||
}
|
||||
|
||||
// binPath returns the path to the file storing the binary data.
|
||||
@@ -339,6 +344,22 @@ func (session *DecomposedFsSession) ScanData() (string, time.Time) {
|
||||
return session.info.MetaData["scanResult"], d
|
||||
}
|
||||
|
||||
// SetStatus sets the status of the upload session
|
||||
func (session *DecomposedFsSession) SetStatus(status, msg string) {
|
||||
session.info.MetaData["status"] = status
|
||||
session.info.MetaData["statusMessage"] = msg
|
||||
}
|
||||
|
||||
// Status returns the status of the upload session
|
||||
func (session *DecomposedFsSession) Status() string {
|
||||
return session.info.MetaData["status"]
|
||||
}
|
||||
|
||||
// StatusMessage returns the status message of the upload session
|
||||
func (session *DecomposedFsSession) StatusMessage() string {
|
||||
return session.info.MetaData["statusMessage"]
|
||||
}
|
||||
|
||||
// sessionPath returns the path to the .info file storing the file's info.
|
||||
func sessionPath(root, id string) string {
|
||||
return filepath.Join(root, "uploads", id+".info")
|
||||
|
||||
Generated
Vendored
+3
-1
@@ -94,7 +94,9 @@ func (store DecomposedFsStore) New(ctx context.Context) *DecomposedFsSession {
|
||||
Storage: map[string]string{
|
||||
"Type": "DecomposedFsStore",
|
||||
},
|
||||
MetaData: tusd.MetaData{},
|
||||
MetaData: tusd.MetaData{
|
||||
"status": string(SessionStatusUploading),
|
||||
},
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
Generated
Vendored
+10
@@ -194,8 +194,18 @@ func (session *DecomposedFsSession) FinishUploadDecomposed(ctx context.Context)
|
||||
|
||||
n, err := session.store.CreateNodeForUpload(ctx, session, attrs)
|
||||
if err != nil {
|
||||
session.SetStatus(SessionStatusFailed, err.Error())
|
||||
if perr := session.Persist(ctx); perr != nil {
|
||||
log.Error().Err(perr).Msg("failed to persist upload session after setting status to failed")
|
||||
}
|
||||
return err
|
||||
}
|
||||
|
||||
session.SetStatus(SessionStatusProcessing, "")
|
||||
if err = session.Persist(ctx); err != nil {
|
||||
log.Error().Err(err).Msg("failed to persist upload session after setting status to processing")
|
||||
}
|
||||
|
||||
// increase the processing counter for every started processing
|
||||
// will be decreased in Cleanup()
|
||||
metrics.UploadProcessing.Inc()
|
||||
|
||||
+5
@@ -94,6 +94,11 @@ type UploadSession interface {
|
||||
|
||||
// ScanData returns the scan data for the UploadSession
|
||||
ScanData() (string, time.Time)
|
||||
|
||||
// Status returns the status for the UploadSession
|
||||
Status() string
|
||||
// StatusMessage returns the status message for the UploadSession
|
||||
StatusMessage() string
|
||||
}
|
||||
|
||||
// UploadSessionFilter can be used to filter upload sessions
|
||||
|
||||
Generated
Vendored
+16
@@ -338,6 +338,22 @@ func (s *OcisSession) ScanData() (string, time.Time) {
|
||||
return s.info.MetaData["scanResult"], d
|
||||
}
|
||||
|
||||
func (s *OcisSession) SetStatus(status string) {
|
||||
s.info.MetaData["status"] = status
|
||||
}
|
||||
|
||||
func (s *OcisSession) Status() string {
|
||||
return s.info.MetaData["status"]
|
||||
}
|
||||
|
||||
func (s *OcisSession) SetStatusMessage(message string) {
|
||||
s.info.MetaData["status_message"] = message
|
||||
}
|
||||
|
||||
func (s *OcisSession) StatusMessage() string {
|
||||
return s.info.MetaData["status_message"]
|
||||
}
|
||||
|
||||
// sessionPath returns the path to the .info file storing the file's info.
|
||||
func sessionPath(root, id string) string {
|
||||
return filepath.Join(root, "uploads", id+".info")
|
||||
|
||||
+87
-4
@@ -23,10 +23,12 @@ import (
|
||||
"errors"
|
||||
"fmt"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
identityUser "github.com/cs3org/go-cs3apis/cs3/identity/user/v1beta1"
|
||||
"github.com/go-ldap/ldap/v3"
|
||||
"github.com/google/uuid"
|
||||
"github.com/hashicorp/golang-lru/v2/expirable"
|
||||
"github.com/opencloud-eu/reva/v2/pkg/appctx"
|
||||
"github.com/opencloud-eu/reva/v2/pkg/errtypes"
|
||||
"github.com/opencloud-eu/reva/v2/pkg/sharedconf"
|
||||
@@ -38,12 +40,45 @@ import (
|
||||
|
||||
// Identity provides methods to query users and groups from an LDAP server
|
||||
type Identity struct {
|
||||
User userConfig `mapstructure:",squash"`
|
||||
Group groupConfig `mapstructure:",squash"`
|
||||
Tenant tenantConfig `mapstructure:",squash"`
|
||||
User userConfig `mapstructure:",squash"`
|
||||
Group groupConfig `mapstructure:",squash"`
|
||||
Tenant tenantConfig `mapstructure:",squash"`
|
||||
LookupCacheTTL string `mapstructure:"lookup_cache_ttl"`
|
||||
|
||||
lookupCache entryCache
|
||||
}
|
||||
|
||||
const tracerName = "pkg/utils/ldap"
|
||||
// entryCache is a small nil-safe wrapper around the expirable LRU used to cache
|
||||
// LDAP lookups. A zero-value entryCache (nil lru) is a valid, disabled cache:
|
||||
// Get always misses and Add is a no-op. This lets the cache be turned off
|
||||
// entirely (e.g. in acceptance tests) by configuring a lookup_cache_ttl of 0.
|
||||
type entryCache struct {
|
||||
lru *expirable.LRU[string, *ldap.Entry]
|
||||
}
|
||||
|
||||
// Get returns the cached entry for key. It reports a miss when the cache is
|
||||
// disabled (nil lru).
|
||||
func (c entryCache) Get(key string) (*ldap.Entry, bool) {
|
||||
if c.lru == nil {
|
||||
return nil, false
|
||||
}
|
||||
return c.lru.Get(key)
|
||||
}
|
||||
|
||||
// Add stores entry under key. It is a no-op when the cache is disabled (nil lru).
|
||||
func (c entryCache) Add(key string, entry *ldap.Entry) {
|
||||
if c.lru == nil {
|
||||
return
|
||||
}
|
||||
c.lru.Add(key, entry)
|
||||
}
|
||||
|
||||
const (
|
||||
tracerName = "pkg/utils/ldap"
|
||||
|
||||
lookupCacheDefaultTTL = 10 * time.Second
|
||||
lookupCacheSize = 1024
|
||||
)
|
||||
|
||||
type userConfig struct {
|
||||
BaseDN string `mapstructure:"user_base_dn"`
|
||||
@@ -223,6 +258,27 @@ func (i *Identity) Setup() error {
|
||||
}
|
||||
}
|
||||
|
||||
if i.LookupCacheTTL != "" {
|
||||
// Parse the TTL string if provided
|
||||
parsedTTL, err := time.ParseDuration(i.LookupCacheTTL)
|
||||
if err != nil {
|
||||
return fmt.Errorf("error parsing lookup_cache_ttl %q: %w", i.LookupCacheTTL, err)
|
||||
}
|
||||
switch {
|
||||
case parsedTTL < 0:
|
||||
return fmt.Errorf("error configuring lookup cache ttl: duration must be >= 0")
|
||||
case parsedTTL == 0:
|
||||
// A TTL of 0 disables the lookup cache entirely. A zero-value
|
||||
// entryCache always misses and never stores anything.
|
||||
i.lookupCache = entryCache{}
|
||||
default:
|
||||
i.lookupCache = entryCache{lru: expirable.NewLRU[string, *ldap.Entry](lookupCacheSize, nil, parsedTTL)}
|
||||
}
|
||||
} else {
|
||||
// Use default TTL if not provided
|
||||
i.lookupCache = entryCache{lru: expirable.NewLRU[string, *ldap.Entry](lookupCacheSize, nil, lookupCacheDefaultTTL)}
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
@@ -254,6 +310,12 @@ func (i *Identity) GetLDAPUserByFilter(ctx context.Context, lc ldap.Client, filt
|
||||
log := appctx.GetLogger(ctx)
|
||||
_, span := appctx.GetTracerProvider(ctx).Tracer(tracerName).Start(ctx, "GetLDAPUserByFilter")
|
||||
defer span.End()
|
||||
|
||||
cacheKey := fmt.Sprintf("user:filter:%s", filter)
|
||||
if cached, ok := i.lookupCache.Get(cacheKey); ok {
|
||||
return cached, nil
|
||||
}
|
||||
|
||||
searchRequest := ldap.NewSearchRequest(
|
||||
i.User.BaseDN, i.User.scopeVal, ldap.NeverDerefAliases, 1, 0, false,
|
||||
filter,
|
||||
@@ -275,6 +337,7 @@ func (i *Identity) GetLDAPUserByFilter(ctx context.Context, lc ldap.Client, filt
|
||||
return nil, errtypes.NotFound(filter)
|
||||
}
|
||||
span.SetStatus(codes.Ok, "")
|
||||
i.lookupCache.Add(cacheKey, res.Entries[0])
|
||||
|
||||
return res.Entries[0], nil
|
||||
}
|
||||
@@ -286,6 +349,11 @@ func (i *Identity) GetLDAPUserByDN(ctx context.Context, lc ldap.Client, dn strin
|
||||
_, span := appctx.GetTracerProvider(ctx).Tracer(tracerName).Start(ctx, "GetLDAPUserByDN")
|
||||
defer span.End()
|
||||
|
||||
cacheKey := fmt.Sprintf("user:dn:%s", dn)
|
||||
if cached, ok := i.lookupCache.Get(cacheKey); ok {
|
||||
return cached, nil
|
||||
}
|
||||
|
||||
filter := fmt.Sprintf("(objectclass=%s)", i.User.Objectclass)
|
||||
if i.User.Filter != "" {
|
||||
filter = fmt.Sprintf("(&%s%s)", i.User.Filter, filter)
|
||||
@@ -310,6 +378,7 @@ func (i *Identity) GetLDAPUserByDN(ctx context.Context, lc ldap.Client, dn strin
|
||||
if len(res.Entries) == 0 {
|
||||
return nil, errtypes.NotFound(dn)
|
||||
}
|
||||
i.lookupCache.Add(cacheKey, res.Entries[0])
|
||||
|
||||
return res.Entries[0], nil
|
||||
}
|
||||
@@ -483,6 +552,12 @@ func (i *Identity) GetLDAPGroupByFilter(ctx context.Context, lc ldap.Client, fil
|
||||
ctx, span := appctx.GetTracerProvider(ctx).Tracer(tracerName).Start(ctx, "GetLDAPGroupByFilter")
|
||||
defer span.End()
|
||||
log := appctx.GetLogger(ctx)
|
||||
|
||||
cacheKey := fmt.Sprintf("group:filter:%s", filter)
|
||||
if cached, ok := i.lookupCache.Get(cacheKey); ok {
|
||||
return cached, nil
|
||||
}
|
||||
|
||||
searchRequest := ldap.NewSearchRequest(
|
||||
i.Group.BaseDN, i.Group.scopeVal, ldap.NeverDerefAliases, 1, 0, false,
|
||||
filter,
|
||||
@@ -511,6 +586,7 @@ func (i *Identity) GetLDAPGroupByFilter(ctx context.Context, lc ldap.Client, fil
|
||||
return nil, errtypes.NotFound(filter)
|
||||
}
|
||||
span.SetStatus(codes.Ok, "")
|
||||
i.lookupCache.Add(cacheKey, res.Entries[0])
|
||||
return res.Entries[0], nil
|
||||
}
|
||||
|
||||
@@ -901,6 +977,12 @@ func (i *Identity) GetLDAPTenantByFilter(ctx context.Context, lc ldap.Client, fi
|
||||
log := appctx.GetLogger(ctx)
|
||||
_, span := appctx.GetTracerProvider(ctx).Tracer(tracerName).Start(ctx, "GetLDAPTenantByFilter")
|
||||
defer span.End()
|
||||
|
||||
cacheKey := fmt.Sprintf("tenant:filter:%s", filter)
|
||||
if cached, ok := i.lookupCache.Get(cacheKey); ok {
|
||||
return cached, nil
|
||||
}
|
||||
|
||||
searchRequest := ldap.NewSearchRequest(
|
||||
i.Tenant.BaseDN, i.Tenant.scopeVal, ldap.NeverDerefAliases, 1, 0, false,
|
||||
filter,
|
||||
@@ -922,6 +1004,7 @@ func (i *Identity) GetLDAPTenantByFilter(ctx context.Context, lc ldap.Client, fi
|
||||
return nil, errtypes.NotFound(filter)
|
||||
}
|
||||
span.SetStatus(codes.Ok, "")
|
||||
i.lookupCache.Add(cacheKey, res.Entries[0])
|
||||
|
||||
return res.Entries[0], nil
|
||||
}
|
||||
|
||||
Reference in new issue
Block a user