fix(collaboration): check the requester's access before a mention goes out

services/collaboration/pkg/notification/service.go

@@ -1,7 +1,6 @@
 package notification
 
 import (
-	"context"
 	"encoding/json"
 	"io"
 	"net/http"
@@ -69,13 +68,15 @@ func NewService(options ServiceOptions) (Service, error) {
 }
 
 func (s Service) HandleNotification(w http.ResponseWriter, r *http.Request) {
+	ctx := metadata.AppendToOutgoingContext(r.Context(), revactx.TokenHeader, r.Header.Get(revactx.TokenHeader))
+
 	gatewayClient, err := s.gatewaySelector.Next()
 	if err != nil {
 		w.WriteHeader(http.StatusInternalServerError)
 		return
 	}
 
-	requestUser, canManage, err := collaboration.CheckPermissions(gatewayClient, r.Context(), collaboration.PermissionCollaborationPublishNotification)
+	requestUser, canManage, err := collaboration.CheckPermissions(gatewayClient, ctx, collaboration.PermissionCollaborationPublishNotification)
 	switch {
 	case err != nil:
 		w.WriteHeader(http.StatusInternalServerError)
@@ -88,60 +89,81 @@ func (s Service) HandleNotification(w http.ResponseWriter, r *http.Request) {
 	defer func() { _ = r.Body.Close() }()
 	body, err := io.ReadAll(r.Body)
 	if err != nil {
-		w.WriteHeader(http.StatusInternalServerError)
+		w.WriteHeader(http.StatusBadRequest)
 		return
 	}
 
 	var data = struct {
-		Type    string   `json:"type" validate:"required"`
-		UserIDs []string `json:"userIDs" validate:"required"`
+		UserIDs []string `json:"userIDs" validate:"required,min=1"`
 		FileID  string   `json:"fileID" validate:"required"`
 	}{}
 	if err := json.Unmarshal(body, &data); err != nil {
-		w.WriteHeader(http.StatusInternalServerError)
+		w.WriteHeader(http.StatusBadRequest)
 		return
 	}
 
 	if err := validate.Struct(data); err != nil {
+		w.WriteHeader(http.StatusBadRequest)
+		return
+	}
+
+	resourceID, err := storagespace.ParseID(data.FileID)
+	if err != nil {
+		w.WriteHeader(http.StatusBadRequest)
+		return
+	}
+
+	ref := &storageprovider.Reference{ResourceId: &resourceID}
+
+	// accept the mention only if the requester can see the file himself
+	statResponse, err := gatewayClient.Stat(ctx, &storageprovider.StatRequest{Ref: ref})
+	switch {
+	case err != nil:
 		w.WriteHeader(http.StatusInternalServerError)
 		return
+	case statResponse.GetStatus().GetCode() != rpcv1beta1.Code_CODE_OK:
+		w.WriteHeader(http.StatusNotFound)
+		return
 	}
 
 	event := ocEvents.ResourceMention{
 		Executant: requestUser.GetId(),
+		Ref:       &storageprovider.Reference{ResourceId: statResponse.GetInfo().GetId()},
 		Timestamp: time.Now(),
 	}
 
+	seen := make(map[string]struct{}, len(data.UserIDs))
 	for _, userID := range data.UserIDs {
-		authResponse, err := gatewayClient.Authenticate(context.Background(), &gateway.AuthenticateRequest{
+		if _, ok := seen[userID]; ok {
+			continue
+		}
+		seen[userID] = struct{}{}
+
+		authResponse, err := gatewayClient.Authenticate(r.Context(), &gateway.AuthenticateRequest{
 			Type:         "machine",
 			ClientId:     "userid:" + userID,
 			ClientSecret: s.machineAuthAPIKey,
 		})
-		if err != nil || authResponse.Status.Code != rpcv1beta1.Code_CODE_OK {
-			w.WriteHeader(http.StatusInternalServerError)
-			return
-		}
-
-		resourceID, err := storagespace.ParseID(data.FileID)
-		if err != nil {
-			w.WriteHeader(http.StatusInternalServerError)
-			return
+		if err != nil || authResponse.GetStatus().GetCode() != rpcv1beta1.Code_CODE_OK {
+			s.log.Debug().Str("userID", userID).Msg("could not authenticate the mentioned user, skipping")
+			continue
 		}
 
 		statResponse, err := gatewayClient.Stat(
-			metadata.AppendToOutgoingContext(context.Background(), revactx.TokenHeader, authResponse.GetToken()),
-			&storageprovider.StatRequest{Ref: &storageprovider.Reference{ResourceId: &resourceID}},
+			metadata.AppendToOutgoingContext(r.Context(), revactx.TokenHeader, authResponse.GetToken()),
+			&storageprovider.StatRequest{Ref: ref},
 		)
-		if err != nil || statResponse.Status.Code != rpcv1beta1.Code_CODE_OK {
-			w.WriteHeader(http.StatusInternalServerError)
-			return
+		if err != nil || statResponse.GetStatus().GetCode() != rpcv1beta1.Code_CODE_OK {
+			s.log.Debug().Str("userID", userID).Msg("mentioned user has no access to the resource, skipping")
+			continue
 		}
 
-		event.UserIDs = append(event.UserIDs, authResponse.User.GetId())
-		event.Ref = &storageprovider.Reference{
-			ResourceId: statResponse.GetInfo().GetId(),
-		}
+		event.UserIDs = append(event.UserIDs, authResponse.GetUser().GetId())
+	}
+
+	if len(event.UserIDs) == 0 {
+		w.WriteHeader(http.StatusOK)
+		return
 	}
 
 	if err := events.Publish(r.Context(), s.eventPublisher, event); err != nil {

services/collaboration/pkg/notification/service_test.go

@@ -0,0 +1,197 @@
+package notification_test
+
+import (
+	"context"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"testing"
+
+	gateway "github.com/cs3org/go-cs3apis/cs3/gateway/v1beta1"
+	userpb "github.com/cs3org/go-cs3apis/cs3/identity/user/v1beta1"
+	cs3Permissions "github.com/cs3org/go-cs3apis/cs3/permissions/v1beta1"
+	cs3RPC "github.com/cs3org/go-cs3apis/cs3/rpc/v1beta1"
+	storageprovider "github.com/cs3org/go-cs3apis/cs3/storage/provider/v1beta1"
+	revaCtx "github.com/opencloud-eu/reva/v2/pkg/ctx"
+	revaEvents "github.com/opencloud-eu/reva/v2/pkg/events"
+	"github.com/opencloud-eu/reva/v2/pkg/rgrpc/todo/pool"
+	cs3mocks "github.com/opencloud-eu/reva/v2/tests/cs3mocks/mocks"
+	"github.com/stretchr/testify/mock"
+	"github.com/stretchr/testify/require"
+	"go-micro.dev/v4/events"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/metadata"
+
+	ocEvents "github.com/opencloud-eu/opencloud/pkg/events"
+	"github.com/opencloud-eu/opencloud/pkg/log"
+	"github.com/opencloud-eu/opencloud/services/collaboration/pkg/notification"
+)
+
+const (
+	requesterToken = "requester-token"
+	fileID         = "storage$space!opaque"
+)
+
+type publisher struct {
+	events []any
+}
+
+func (p *publisher) Publish(_ string, event any, _ ...events.PublishOption) error {
+	p.events = append(p.events, event)
+	return nil
+}
+
+func (p *publisher) mentions() []ocEvents.ResourceMention {
+	var mentions []ocEvents.ResourceMention
+	for _, event := range p.events {
+		if mention, ok := event.(ocEvents.ResourceMention); ok {
+			mentions = append(mentions, mention)
+		}
+	}
+
+	return mentions
+}
+
+func token(ctx context.Context) string {
+	md, ok := metadata.FromOutgoingContext(ctx)
+	if !ok {
+		return ""
+	}
+
+	return strings.Join(md.Get(revaCtx.TokenHeader), "")
+}
+
+func statOK() *storageprovider.StatResponse {
+	return &storageprovider.StatResponse{
+		Status: &cs3RPC.Status{Code: cs3RPC.Code_CODE_OK},
+		Info: &storageprovider.ResourceInfo{
+			Id: &storageprovider.ResourceId{StorageId: "storage", SpaceId: "space", OpaqueId: "opaque"},
+		},
+	}
+}
+
+func newService(t *testing.T, p revaEvents.Publisher, canStat ...string) notification.Service {
+	t.Helper()
+
+	gatewayAPIClient := &cs3mocks.GatewayAPIClient{}
+	gatewayAPIClient.On("CheckPermission", mock.Anything, mock.Anything).Return(
+		&cs3Permissions.CheckPermissionResponse{Status: &cs3RPC.Status{Code: cs3RPC.Code_CODE_OK}}, nil)
+
+	gatewayAPIClient.On("Authenticate", mock.Anything, mock.Anything).Return(
+		func(_ context.Context, req *gateway.AuthenticateRequest, _ ...grpc.CallOption) *gateway.AuthenticateResponse {
+			userID := strings.TrimPrefix(req.GetClientId(), "userid:")
+
+			return &gateway.AuthenticateResponse{
+				Status: &cs3RPC.Status{Code: cs3RPC.Code_CODE_OK},
+				Token:  userID + "-token",
+				User:   &userpb.User{Id: &userpb.UserId{OpaqueId: userID}},
+			}
+		}, nil)
+
+	allowed := map[string]struct{}{}
+	for _, name := range canStat {
+		allowed[name+"-token"] = struct{}{}
+	}
+
+	gatewayAPIClient.On("Stat", mock.Anything, mock.Anything).Return(
+		func(ctx context.Context, _ *storageprovider.StatRequest, _ ...grpc.CallOption) *storageprovider.StatResponse {
+			if _, ok := allowed[token(ctx)]; !ok {
+				return &storageprovider.StatResponse{Status: &cs3RPC.Status{Code: cs3RPC.Code_CODE_NOT_FOUND}}
+			}
+
+			return statOK()
+		}, nil)
+
+	svc, err := notification.NewService(
+		notification.ServiceOptions{}.
+			WithLogger(log.NopLogger()).
+			WithEventPublisher(p).
+			WithMachineAuthAPIKey("machine-auth-api-key").
+			WithGatewaySelector(pool.GetSelector[gateway.GatewayAPIClient](
+				"GatewaySelector"+t.Name(),
+				"eu.opencloud.api.gateway",
+				func(cc grpc.ClientConnInterface) gateway.GatewayAPIClient {
+					return gatewayAPIClient
+				},
+			)),
+	)
+	require.NoError(t, err)
+
+	return svc
+}
+
+func newRequest(body string) *http.Request {
+	req := httptest.NewRequest(http.MethodPost, "/collaboration/notify", strings.NewReader(body))
+	req.Header.Set(revaCtx.TokenHeader, requesterToken)
+
+	return req.WithContext(revaCtx.ContextSetUser(req.Context(), &userpb.User{
+		Id: &userpb.UserId{OpaqueId: "requester"},
+	}))
+}
+
+func TestHandleNotification_RequesterWithoutAccess(t *testing.T) {
+	p := &publisher{}
+	svc := newService(t, p, "alice")
+
+	resp := httptest.NewRecorder()
+	svc.HandleNotification(resp, newRequest(`{"userIDs":["alice"],"fileID":"`+fileID+`"}`))
+
+	require.Equal(t, http.StatusNotFound, resp.Code)
+	require.Empty(t, p.mentions())
+}
+
+func TestHandleNotification_SkipsRecipientWithoutAccess(t *testing.T) {
+	p := &publisher{}
+	svc := newService(t, p, "requester", "alice")
+
+	resp := httptest.NewRecorder()
+	svc.HandleNotification(resp, newRequest(`{"userIDs":["alice","carol"],"fileID":"`+fileID+`"}`))
+
+	require.Equal(t, http.StatusOK, resp.Code)
+	require.Len(t, p.mentions(), 1)
+	require.Len(t, p.mentions()[0].UserIDs, 1)
+	require.Equal(t, "alice", p.mentions()[0].UserIDs[0].GetOpaqueId())
+}
+
+func TestHandleNotification_NoRecipientLeft(t *testing.T) {
+	p := &publisher{}
+	svc := newService(t, p, "requester")
+
+	resp := httptest.NewRecorder()
+	svc.HandleNotification(resp, newRequest(`{"userIDs":["alice"],"fileID":"`+fileID+`"}`))
+
+	require.Equal(t, http.StatusOK, resp.Code)
+	require.Empty(t, p.mentions())
+}
+
+func TestHandleNotification_DedupsUserIDs(t *testing.T) {
+	p := &publisher{}
+	svc := newService(t, p, "requester", "alice")
+
+	resp := httptest.NewRecorder()
+	svc.HandleNotification(resp, newRequest(`{"userIDs":["alice","alice","alice"],"fileID":"`+fileID+`"}`))
+
+	require.Equal(t, http.StatusOK, resp.Code)
+	require.Len(t, p.mentions(), 1)
+	require.Len(t, p.mentions()[0].UserIDs, 1)
+}
+
+func TestHandleNotification_BadRequest(t *testing.T) {
+	for name, body := range map[string]string{
+		"no json":       `not json`,
+		"no userIDs":    `{"fileID":"` + fileID + `"}`,
+		"empty userIDs": `{"userIDs":[],"fileID":"` + fileID + `"}`,
+		"no fileID":     `{"userIDs":["alice"]}`,
+	} {
+		t.Run(name, func(t *testing.T) {
+			p := &publisher{}
+			svc := newService(t, p, "requester", "alice")
+
+			resp := httptest.NewRecorder()
+			svc.HandleNotification(resp, newRequest(body))
+
+			require.Equal(t, http.StatusBadRequest, resp.Code)
+			require.Empty(t, p.mentions())
+		})
+	}
+}

services/notifications/pkg/service/resource.go

@@ -26,7 +26,7 @@ func (s eventsNotifier) handleResourceMention(e ocEvents.ResourceMention, eventI
 
 	ctx, err := utils.GetServiceUserContextWithContext(context.Background(), gatewayClient, s.serviceAccountID, s.serviceAccountSecret)
 	if err != nil {
-		logger.Error().Err(err).Msg("could not select next gateway client")
+		logger.Error().Err(err).Msg("could not get service user context")
 		return
 	}
 
@@ -51,7 +51,14 @@ func (s eventsNotifier) handleResourceMention(e ocEvents.ResourceMention, eventI
 			return
 		}
 
+		// the event is not necessarily deduped, a recipient listed twice would be mailed twice
+		seen := make(map[string]struct{}, len(e.UserIDs)+1)
 		for _, userID := range append([]*user.UserId{e.Executant}, e.UserIDs...) {
+			if _, ok := seen[userID.GetOpaqueId()]; ok {
+				continue
+			}
+			seen[userID.GetOpaqueId()] = struct{}{}
+
 			switch u, err := s.getUser(ctx, userID); {
 			case err != nil:
 				logger.Error().Err(err).Msg("could not get user")