diff --git a/services/collaboration/pkg/notification/service.go b/services/collaboration/pkg/notification/service.go index bd1bac80a3..f9a1c1387c 100644 --- a/services/collaboration/pkg/notification/service.go +++ b/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 { diff --git a/services/collaboration/pkg/notification/service_test.go b/services/collaboration/pkg/notification/service_test.go new file mode 100644 index 0000000000..9871a5eafe --- /dev/null +++ b/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()) + }) + } +} diff --git a/services/notifications/pkg/service/resource.go b/services/notifications/pkg/service/resource.go index 6fe13fb23f..1d81bf8814 100644 --- a/services/notifications/pkg/service/resource.go +++ b/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")