Bump reva

vendor/github.com/opencloud-eu/reva/v2/pkg/events/mocks/Stream.go

@@ -40,14 +40,13 @@ func (_m *Stream) EXPECT() *Stream_Expecter {
 
 // Consume provides a mock function with given fields: _a0, _a1
 func (_m *Stream) Consume(_a0 string, _a1 ...events.ConsumeOption) (<-chan events.Event, error) {
-	_va := make([]interface{}, len(_a1))
-	for _i := range _a1 {
-		_va[_i] = _a1[_i]
+	var tmpRet mock.Arguments
+	if len(_a1) > 0 {
+		tmpRet = _m.Called(_a0, _a1)
+	} else {
+		tmpRet = _m.Called(_a0)
 	}
-	var _ca []interface{}
-	_ca = append(_ca, _a0)
-	_ca = append(_ca, _va...)
-	ret := _m.Called(_ca...)
+	ret := tmpRet
 
 	if len(ret) == 0 {
 		panic("no return value specified for Consume")
@@ -113,14 +112,13 @@ func (_c *Stream_Consume_Call) RunAndReturn(run func(string, ...events.ConsumeOp
 
 // Publish provides a mock function with given fields: _a0, _a1, _a2
 func (_m *Stream) Publish(_a0 string, _a1 interface{}, _a2 ...events.PublishOption) error {
-	_va := make([]interface{}, len(_a2))
-	for _i := range _a2 {
-		_va[_i] = _a2[_i]
+	var tmpRet mock.Arguments
+	if len(_a2) > 0 {
+		tmpRet = _m.Called(_a0, _a1, _a2)
+	} else {
+		tmpRet = _m.Called(_a0, _a1)
 	}
-	var _ca []interface{}
-	_ca = append(_ca, _a0, _a1)
-	_ca = append(_ca, _va...)
-	ret := _m.Called(_ca...)
+	ret := tmpRet
 
 	if len(ret) == 0 {
 		panic("no return value specified for Publish")

vendor/github.com/opencloud-eu/reva/v2/pkg/events/raw/certs.go

@@ -0,0 +1,27 @@
+package raw
+
+import (
+	"bytes"
+	"crypto/x509"
+	"errors"
+	"io"
+)
+
+// newCertPoolFromPEM reads certificates from io.Reader and returns a x509.CertPool
+// containing those certificates.
+func newCertPoolFromPEM(crts ...io.Reader) (*x509.CertPool, error) {
+	certPool := x509.NewCertPool()
+
+	var buf bytes.Buffer
+	for _, c := range crts {
+		if _, err := io.Copy(&buf, c); err != nil {
+			return nil, err
+		}
+		if !certPool.AppendCertsFromPEM(buf.Bytes()) {
+			return nil, errors.New("failed to append cert from PEM")
+		}
+		buf.Reset()
+	}
+
+	return certPool, nil
+}

vendor/github.com/opencloud-eu/reva/v2/pkg/events/raw/mocks/Stream.go

@@ -0,0 +1,127 @@
+// 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.
+
+// Code generated by mockery v2.53.2. DO NOT EDIT.
+
+package mocks
+
+import (
+	events "github.com/opencloud-eu/reva/v2/pkg/events"
+	mock "github.com/stretchr/testify/mock"
+
+	raw "github.com/opencloud-eu/reva/v2/pkg/events/raw"
+)
+
+// Stream is an autogenerated mock type for the Stream type
+type Stream struct {
+	mock.Mock
+}
+
+type Stream_Expecter struct {
+	mock *mock.Mock
+}
+
+func (_m *Stream) EXPECT() *Stream_Expecter {
+	return &Stream_Expecter{mock: &_m.Mock}
+}
+
+// Consume provides a mock function with given fields: group, evs
+func (_m *Stream) Consume(group string, evs ...events.Unmarshaller) (<-chan raw.Event, error) {
+	var tmpRet mock.Arguments
+	if len(evs) > 0 {
+		tmpRet = _m.Called(group, evs)
+	} else {
+		tmpRet = _m.Called(group)
+	}
+	ret := tmpRet
+
+	if len(ret) == 0 {
+		panic("no return value specified for Consume")
+	}
+
+	var r0 <-chan raw.Event
+	var r1 error
+	if rf, ok := ret.Get(0).(func(string, ...events.Unmarshaller) (<-chan raw.Event, error)); ok {
+		return rf(group, evs...)
+	}
+	if rf, ok := ret.Get(0).(func(string, ...events.Unmarshaller) <-chan raw.Event); ok {
+		r0 = rf(group, evs...)
+	} else {
+		if ret.Get(0) != nil {
+			r0 = ret.Get(0).(<-chan raw.Event)
+		}
+	}
+
+	if rf, ok := ret.Get(1).(func(string, ...events.Unmarshaller) error); ok {
+		r1 = rf(group, evs...)
+	} else {
+		r1 = ret.Error(1)
+	}
+
+	return r0, r1
+}
+
+// Stream_Consume_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'Consume'
+type Stream_Consume_Call struct {
+	*mock.Call
+}
+
+// Consume is a helper method to define mock.On call
+//   - group string
+//   - evs ...events.Unmarshaller
+func (_e *Stream_Expecter) Consume(group interface{}, evs ...interface{}) *Stream_Consume_Call {
+	return &Stream_Consume_Call{Call: _e.mock.On("Consume",
+		append([]interface{}{group}, evs...)...)}
+}
+
+func (_c *Stream_Consume_Call) Run(run func(group string, evs ...events.Unmarshaller)) *Stream_Consume_Call {
+	_c.Call.Run(func(args mock.Arguments) {
+		variadicArgs := make([]events.Unmarshaller, len(args)-1)
+		for i, a := range args[1:] {
+			if a != nil {
+				variadicArgs[i] = a.(events.Unmarshaller)
+			}
+		}
+		run(args[0].(string), variadicArgs...)
+	})
+	return _c
+}
+
+func (_c *Stream_Consume_Call) Return(_a0 <-chan raw.Event, _a1 error) *Stream_Consume_Call {
+	_c.Call.Return(_a0, _a1)
+	return _c
+}
+
+func (_c *Stream_Consume_Call) RunAndReturn(run func(string, ...events.Unmarshaller) (<-chan raw.Event, error)) *Stream_Consume_Call {
+	_c.Call.Return(run)
+	return _c
+}
+
+// NewStream creates a new instance of Stream. It also registers a testing interface on the mock and a cleanup function to assert the mocks expectations.
+// The first argument is typically a *testing.T value.
+func NewStream(t interface {
+	mock.TestingT
+	Cleanup(func())
+}) *Stream {
+	mock := &Stream{}
+	mock.Mock.Test(t)
+
+	t.Cleanup(func() { mock.AssertExpectations(t) })
+
+	return mock
+}

vendor/github.com/opencloud-eu/reva/v2/pkg/events/raw/raw.go

@@ -0,0 +1,216 @@
+package raw
+
+import (
+	"context"
+	"crypto/tls"
+	"crypto/x509"
+	"encoding/json"
+	"fmt"
+	"os"
+	"reflect"
+	"time"
+
+	"github.com/cenkalti/backoff"
+	"github.com/nats-io/nats.go"
+	"github.com/nats-io/nats.go/jetstream"
+	"github.com/opencloud-eu/reva/v2/pkg/events"
+	"github.com/pkg/errors"
+)
+
+// Config is the configuration needed for a NATS event stream
+type Config struct {
+	Endpoint             string        `mapstructure:"address"`          // Endpoint of the nats server
+	Cluster              string        `mapstructure:"clusterID"`        // CluserID of the nats cluster
+	TLSInsecure          bool          `mapstructure:"tls-insecure"`     // Whether to verify TLS certificates
+	TLSRootCACertificate string        `mapstructure:"tls-root-ca-cert"` // The root CA certificate used to validate the TLS certificate
+	EnableTLS            bool          `mapstructure:"enable-tls"`       // Enable TLS
+	AuthUsername         string        `mapstructure:"username"`         // Username for authentication
+	AuthPassword         string        `mapstructure:"password"`         // Password for authentication
+	MaxAckPending        int           `mapstructure:"max-ack-pending"`  // Maximum number of unacknowledged messages
+	AckWait              time.Duration `mapstructure:"ack-wait"`         // Time to wait for an ack
+}
+
+type RawEvent struct {
+	Timestamp time.Time
+	Metadata  map[string]string
+	ID        string
+	Topic     string
+	Payload   []byte
+
+	msg jetstream.Msg
+}
+
+type Event struct {
+	events.Event
+
+	msg jetstream.Msg
+}
+
+func (re *Event) Ack() error {
+	if re.msg == nil {
+		return errors.New("cannot ack event without message")
+	}
+	return re.msg.Ack()
+}
+
+func (re *Event) InProgress() error {
+	if re.msg == nil {
+		return errors.New("cannot mark event as in progress without message")
+	}
+	return re.msg.InProgress()
+}
+
+type Stream interface {
+	Consume(group string, evs ...events.Unmarshaller) (<-chan Event, error)
+}
+
+type RawStream struct {
+	Js jetstream.Stream
+
+	c Config
+}
+
+func FromConfig(ctx context.Context, name string, cfg Config) (Stream, error) {
+	var s Stream
+	b := backoff.NewExponentialBackOff()
+
+	connect := func() error {
+		var tlsConf *tls.Config
+		if cfg.EnableTLS {
+			var rootCAPool *x509.CertPool
+			if cfg.TLSRootCACertificate != "" {
+				rootCrtFile, err := os.Open(cfg.TLSRootCACertificate)
+				if err != nil {
+					return err
+				}
+
+				rootCAPool, err = newCertPoolFromPEM(rootCrtFile)
+				if err != nil {
+					return err
+				}
+				cfg.TLSInsecure = false
+			}
+
+			tlsConf = &tls.Config{
+				MinVersion:         tls.VersionTLS12,
+				InsecureSkipVerify: cfg.TLSInsecure,
+				RootCAs:            rootCAPool,
+			}
+		}
+
+		nopts := nats.GetDefaultOptions()
+		nopts.Name = name
+		if tlsConf != nil {
+			nopts.Secure = true
+			nopts.TLSConfig = tlsConf
+		}
+
+		if len(cfg.Endpoint) > 0 {
+			nopts.Servers = []string{cfg.Endpoint}
+		}
+
+		if cfg.AuthUsername != "" && cfg.AuthPassword != "" {
+			nopts.User = cfg.AuthUsername
+			nopts.Password = cfg.AuthPassword
+		}
+
+		conn, err := nopts.Connect()
+		if err != nil {
+			return err
+		}
+
+		jsConn, err := jetstream.New(conn)
+		if err != nil {
+			return err
+		}
+
+		js, err := jsConn.Stream(ctx, events.MainQueueName)
+		if err != nil {
+			return err
+		}
+
+		s = &RawStream{
+			Js: js,
+			c:  cfg,
+		}
+		return nil
+	}
+	err := backoff.Retry(connect, b)
+	if err != nil {
+		return s, errors.Wrap(err, "could not connect to nats jetstream")
+	}
+	return s, nil
+}
+
+func (s *RawStream) Consume(group string, evs ...events.Unmarshaller) (<-chan Event, error) {
+	c, err := s.consumeRaw(group)
+	if err != nil {
+		return nil, err
+	}
+
+	registeredEvents := map[string]events.Unmarshaller{}
+	for _, e := range evs {
+		typ := reflect.TypeOf(e)
+		registeredEvents[typ.String()] = e
+	}
+
+	outchan := make(chan Event)
+	go func() {
+		for {
+			e := <-c
+			eventType := e.Metadata[events.MetadatakeyEventType]
+			ev, ok := registeredEvents[eventType]
+			if !ok {
+				_ = e.msg.Ack() // Discard. We are not interested in this event type
+				continue
+			}
+
+			event, err := ev.Unmarshal(e.Payload)
+			if err != nil {
+				continue
+			}
+
+			outchan <- Event{
+				Event: events.Event{
+					Type:        eventType,
+					ID:          e.Metadata[events.MetadatakeyEventID],
+					TraceParent: e.Metadata[events.MetadatakeyTraceParent],
+					InitiatorID: e.Metadata[events.MetadatakeyInitiatorID],
+					Event:       event,
+				},
+				msg: e.msg,
+			}
+		}
+	}()
+	return outchan, nil
+}
+
+func (s *RawStream) consumeRaw(group string) (<-chan RawEvent, error) {
+	consumer, err := s.Js.CreateOrUpdateConsumer(context.Background(), jetstream.ConsumerConfig{
+		Durable:       group,
+		DeliverPolicy: jetstream.DeliverNewPolicy,
+		AckPolicy:     jetstream.AckExplicitPolicy, // Require manual acknowledgment
+		MaxAckPending: s.c.MaxAckPending,           // Maximum number of unacknowledged messages
+		AckWait:       s.c.AckWait,                 // Time to wait for an ack
+	})
+	if err != nil {
+		return nil, err
+	}
+
+	channel := make(chan RawEvent)
+	callback := func(msg jetstream.Msg) {
+		var rawEvent RawEvent
+		if err := json.Unmarshal(msg.Data(), &rawEvent); err != nil {
+			fmt.Printf("error unmarshalling event: %v\n", err)
+			return
+		}
+		rawEvent.msg = msg
+		channel <- rawEvent
+	}
+	_, err = consumer.Consume(callback)
+	if err != nil {
+		return nil, err
+	}
+
+	return channel, nil
+}

vendor/github.com/opencloud-eu/reva/v2/pkg/micro/ocdav/option.go

@@ -401,3 +401,10 @@ func RegisterInterval(interval time.Duration) Option {
 		o.RegisterInterval = interval
 	}
 }
+
+// URLSigningSharedSecret provides a function to set the URLSigningSharedSecret config option.
+func URLSigningSharedSecret(secret string) Option {
+	return func(o *Options) {
+		o.config.URLSigningSharedSecret = secret
+	}
+}

vendor/github.com/opencloud-eu/reva/v2/pkg/signedurl/jwt.go

@@ -0,0 +1,116 @@
+package signedurl
+
+import (
+	"errors"
+	"fmt"
+	"net/url"
+	"time"
+
+	"github.com/golang-jwt/jwt/v5"
+)
+
+// JWTSignedURL implements the Signer and Verifier interfaces using JWT for signing URLs.
+type JWTSignedURL struct {
+	JWTOptions
+}
+
+type claims struct {
+	TargetURL string `json:"target_url"`
+	jwt.RegisteredClaims
+}
+
+// JWTOption defines a single option function.
+type JWTOption func(o *JWTOptions)
+
+// JWTOptions defines the available options for this package.
+type JWTOptions struct {
+	secret     string // Secret key used for signing and verifying JWTs
+	queryParam string // Name of the query parameter for the signature
+}
+
+func NewJWTSignedURL(opts ...JWTOption) (*JWTSignedURL, error) {
+	opt := JWTOptions{}
+	for _, o := range opts {
+		o(&opt)
+	}
+
+	if opt.secret == "" {
+		return nil, ErrInvalidKey
+	}
+
+	if opt.queryParam == "" {
+		opt.queryParam = "oc-jwt-sig"
+	}
+
+	return &JWTSignedURL{opt}, nil
+}
+
+func WithSecret(secret string) JWTOption {
+	return func(o *JWTOptions) {
+		o.secret = secret
+	}
+}
+
+func WithQueryParam(queryParam string) JWTOption {
+	return func(o *JWTOptions) {
+		o.queryParam = queryParam
+	}
+}
+
+// Sign signs a URL using JWT with a specified time-to-live (ttl).
+func (j *JWTSignedURL) Sign(unsignedURL, subject string, ttl time.Duration) (string, error) {
+	// Re-encode the Query parameters to ensure they are "normalized" (Values.Encode() does return them alphabetically ordered).
+	u, err := url.Parse(unsignedURL)
+	if err != nil {
+		return "", NewSignedURLError(err, "failed to parse url")
+	}
+	query := u.Query()
+	u.RawQuery = query.Encode()
+	c := claims{
+		TargetURL: u.String(),
+		RegisteredClaims: jwt.RegisteredClaims{
+			ExpiresAt: jwt.NewNumericDate(time.Now().Add(ttl)),
+			Issuer:    "reva",
+			IssuedAt:  jwt.NewNumericDate(time.Now()),
+			Subject:   subject,
+		},
+	}
+	token := jwt.NewWithClaims(jwt.SigningMethodHS256, c)
+	signedToken, err := token.SignedString([]byte(j.secret))
+	if err != nil {
+		return "", fmt.Errorf("signing failed: %w", err)
+	}
+	query.Set(j.queryParam, signedToken)
+	u.RawQuery = query.Encode()
+	return u.String(), nil
+}
+
+// Verify verifies a signed URL using a JWT. Returns the subject of the JWT if verification is successful.
+func (j *JWTSignedURL) Verify(signedURL string) (string, error) {
+	u, err := url.Parse(signedURL)
+	if err != nil {
+		return "", NewSignatureVerificationError(fmt.Errorf("could not parse URL: %w", err))
+	}
+	query := u.Query()
+	tokenString := query.Get(j.queryParam)
+	if tokenString == "" {
+		return "", NewSignatureVerificationError(errors.New("no signature in url"))
+	}
+	token, err := jwt.ParseWithClaims(tokenString, &claims{}, func(token *jwt.Token) (any, error) { return []byte(j.secret), nil })
+	if err != nil {
+		return "", NewSignatureVerificationError(err)
+	}
+	c, ok := token.Claims.(*claims)
+	if !ok {
+		return "", NewSignatureVerificationError(errors.New("invalid JWT claims"))
+	}
+
+	query.Del(j.queryParam)
+	u.RawQuery = query.Encode()
+
+	if c.TargetURL != u.String() {
+		return "", NewSignatureVerificationError(errors.New("url mismatch"))
+	}
+
+	return c.Subject, nil
+}

vendor/github.com/opencloud-eu/reva/v2/pkg/signedurl/signedurl.go

@@ -0,0 +1,64 @@
+// Package signedurl provides interfaces and implementations for signing and verifying URLs.
+package signedurl
+
+import (
+	"time"
+)
+
+type Signer interface {
+	// Sign signs a URL
+	Sign(url, principal string, ttl time.Duration) (string, error)
+}
+
+type Verifier interface {
+	// Verify verifies a signed URL
+	Verify(signedURL string) (string, error)
+}
+
+type SignedURLError struct {
+	innerErr error
+	message  string
+}
+
+// NewSignedURLError creates a new SignedURLError with the provided inner error and message.
+func NewSignedURLError(innerErr error, message string) SignedURLError {
+	return SignedURLError{
+		innerErr: innerErr,
+		message:  message,
+	}
+}
+
+var ErrInvalidKey = NewSignedURLError(nil, "invalid key provided")
+
+type SignatureVerificationError struct {
+	SignedURLError
+}
+
+func NewSignatureVerificationError(innerErr error) SignatureVerificationError {
+	return SignatureVerificationError{
+		SignedURLError: SignedURLError{
+			innerErr: innerErr,
+			message:  "signature verification failed",
+		},
+	}
+}
+
+func (e SignatureVerificationError) Is(tgt error) bool {
+	// Check if the target error is of type SignatureVerificationError
+	if _, ok := tgt.(SignatureVerificationError); ok {
+		return true
+	}
+	return false
+}
+
+// Error implements the error interface for errorConst.
+func (e SignedURLError) Error() string {
+	if e.innerErr != nil {
+		return e.message + ": " + e.innerErr.Error()
+	}
+	return e.message
+}
+
+func (e SignedURLError) Unwrap() error {
+	return e.innerErr
+}

vendor/github.com/opencloud-eu/reva/v2/pkg/storage/fs/posix/tree/cephfswatcher.go

@@ -98,11 +98,13 @@ func (w *CephFSWatcher) Watch(topic string) {
 			switch {
 			case mask&CEPH_MDS_NOTIFY_DELETE > 0:
 				err = w.tree.Scan(path, ActionDelete, isDir)
-			case mask&CEPH_MDS_NOTIFY_CREATE > 0 || mask&CEPH_MDS_NOTIFY_MOVED_TO > 0:
+			case mask&CEPH_MDS_NOTIFY_MOVED_TO > 0:
 				if ev.SrcMask > 0 {
 					// This is a move, clean up the old path
 					err = w.tree.Scan(filepath.Join(w.tree.options.WatchRoot, ev.SrcPath), ActionMoveFrom, isDir)
 				}
+				err = w.tree.Scan(path, ActionMove, isDir)
+			case mask&CEPH_MDS_NOTIFY_CREATE > 0:
 				err = w.tree.Scan(path, ActionCreate, isDir)
 			case mask&CEPH_MDS_NOTIFY_CLOSE_WRITE > 0:
 				err = w.tree.Scan(path, ActionUpdate, isDir)

vendor/github.com/opencloud-eu/reva/v2/pkg/storage/fs/posix/tree/inotifywatcher.go

@@ -92,7 +92,9 @@ func (iw *InotifyWatcher) Watch(path string) {
 						err = iw.tree.Scan(event.Filename, ActionDelete, event.IsDir)
 					case inotifywaitgo.MOVED_FROM:
 						err = iw.tree.Scan(event.Filename, ActionMoveFrom, event.IsDir)
-					case inotifywaitgo.CREATE, inotifywaitgo.MOVED_TO:
+					case inotifywaitgo.MOVED_TO:
+						err = iw.tree.Scan(event.Filename, ActionMove, event.IsDir)
+					case inotifywaitgo.CREATE:
 						err = iw.tree.Scan(event.Filename, ActionCreate, event.IsDir)
 					case inotifywaitgo.CLOSE_WRITE:
 						err = iw.tree.Scan(event.Filename, ActionUpdate, event.IsDir)

vendor/github.com/opencloud-eu/reva/v2/pkg/storage/pkg/decomposedfs/node/node.go

@@ -1117,22 +1117,22 @@ func (n *Node) ReadUserPermissions(ctx context.Context, u *userpb.User) (ap *pro
 			continue
 		}
 
-		if isGrantExpired(g) {
-			continue
-		}
-
 		switch {
 		case err == nil:
+			if isGrantExpired(g) {
+				continue
+			}
+
 			// If all permissions are set to false we have a deny grant
 			if grants.PermissionsEqual(g.Permissions, &provider.ResourcePermissions{}) {
 				return NoPermissions(), true, nil
 			}
 			AddPermissions(ap, g.GetPermissions())
 		case metadata.IsAttrUnset(err):
-			appctx.GetLogger(ctx).Error().Str("spaceid", n.SpaceID).Str("nodeid", n.ID).Str("grant", grantees[i]).Interface("grantees", grantees).Msg("grant vanished from node after listing")
+			appctx.GetLogger(ctx).Error().Err(err).Str("spaceid", n.SpaceID).Str("nodeid", n.ID).Str("path", n.InternalPath()).Str("grant", grantees[i]).Interface("grantees", grantees).Msg("grant vanished from node after listing")
 			// continue with next segment
 		default:
-			appctx.GetLogger(ctx).Error().Err(err).Str("spaceid", n.SpaceID).Str("nodeid", n.ID).Str("grant", grantees[i]).Msg("error reading permissions")
+			appctx.GetLogger(ctx).Error().Err(err).Str("spaceid", n.SpaceID).Str("nodeid", n.ID).Str("path", n.InternalPath()).Str("grant", grantees[i]).Msg("error reading permissions")
 			// continue with next segment
 		}
 	}