Merge pull request #3297 from fschade/fix-mention-notification-review

fix(collaboration): harden the mention notification endpoint

services/collaboration/pkg/collaboration/collaboration.go

@@ -14,8 +14,7 @@ import (
 type Permission string
 
 const (
-	PermissionCollaborationManageFonts         Permission = "Collaboration.Fonts.Manage"
-	PermissionCollaborationPublishNotification Permission = "Collaboration.Notification.Publish"
+	PermissionCollaborationManageFonts Permission = "Collaboration.Fonts.Manage"
 )
 
 func CheckPermissions(gatewayClient gateway.GatewayAPIClient, ctx context.Context, permission Permission) (*userpb.User, bool, error) {

services/collaboration/pkg/command/server.go

@@ -8,7 +8,6 @@ import (
 	"os/signal"
 	"time"
 
-	"github.com/opencloud-eu/reva/v2/pkg/events/stream"
 	"github.com/opencloud-eu/reva/v2/pkg/rgrpc/todo/pool"
 	"github.com/opencloud-eu/reva/v2/pkg/store"
 	"github.com/spf13/afero"
@@ -17,7 +16,6 @@ import (
 	microstore "go-micro.dev/v4/store"
 
 	"github.com/opencloud-eu/opencloud/pkg/config/configlog"
-	"github.com/opencloud-eu/opencloud/pkg/generators"
 	"github.com/opencloud-eu/opencloud/pkg/log"
 	"github.com/opencloud-eu/opencloud/pkg/registry"
 	"github.com/opencloud-eu/opencloud/pkg/runner"
@@ -28,7 +26,6 @@ import (
 	"github.com/opencloud-eu/opencloud/services/collaboration/pkg/connector"
 	"github.com/opencloud-eu/opencloud/services/collaboration/pkg/font"
 	"github.com/opencloud-eu/opencloud/services/collaboration/pkg/helpers"
-	"github.com/opencloud-eu/opencloud/services/collaboration/pkg/notification"
 	"github.com/opencloud-eu/opencloud/services/collaboration/pkg/server/debug"
 	"github.com/opencloud-eu/opencloud/services/collaboration/pkg/server/grpc"
 	"github.com/opencloud-eu/opencloud/services/collaboration/pkg/server/http"
@@ -174,32 +171,8 @@ func Server(cfg *config.Config) *cobra.Command {
 				}
 			}
 
-			var optionalHTTPServerOptions []http.Option
-			var notificationService notification.Service
-			if cfg.Events.Endpoint == "" {
-				logger.Warn().Msg("Events endpoint is not configured, notifications from the collaboration service will not work")
-			} else {
-				connName := generators.GenerateConnectionName(cfg.Service.Name, generators.NTypeBus)
-				natsStream, err := stream.NatsFromConfig(connName, true, stream.NatsConfig(cfg.Events))
-				if err != nil {
-					return err
-				}
-				notificationService, err = notification.NewService(
-					notification.ServiceOptions{}.
-						WithLogger(logger).
-						WithGatewaySelector(gatewaySelector).
-						WithEventPublisher(natsStream).
-						WithMachineAuthAPIKey(cfg.MachineAuthAPIKey),
-				)
-				if err != nil {
-					return err
-				}
-
-				optionalHTTPServerOptions = append(optionalHTTPServerOptions, http.NotificationService(&notificationService))
-			}
-
 			// start HTTP server
-			httpServer, err := http.Server(append([]http.Option{
+			httpServer, err := http.Server(
 				http.Adapter(connector.NewHttpAdapter(gatewaySelector, cfg, st, selector.NewSelector(selector.Registry(registry.GetRegistry())))),
 				http.Logger(logger),
 				http.Config(cfg),
@@ -207,7 +180,7 @@ func Server(cfg *config.Config) *cobra.Command {
 				http.TracerProvider(traceProvider),
 				http.Store(st),
 				http.FontService(fontService),
-			}, optionalHTTPServerOptions...)...)
+			)
 			if err != nil {
 				logger.Info().Err(err).Str("transport", "http").Msg("Failed to initialize server")
 				return err

services/collaboration/pkg/config/config.go

@@ -14,7 +14,6 @@ type Config struct {
 	App     App     `yaml:"app"`
 	Font    Font    `yaml:"font"`
 	Store   Store   `yaml:"store"`
-	Events  Events  `yaml:"events"`
 
 	TokenManager *TokenManager `yaml:"token_manager"`
 
@@ -28,6 +27,4 @@ type Config struct {
 	Debug    Debug  `yaml:"debug"`
 
 	Context context.Context `yaml:"-"`
-
-	MachineAuthAPIKey string `yaml:"machine_auth_api_key" env:"OC_MACHINE_AUTH_API_KEY;COLLABORATION_MACHINE_AUTH_API_KEY" desc:"The machine auth API key used to validate internal requests necessary to access resources from other services." introductionVersion:"7.3.0"`
 }

services/collaboration/pkg/config/defaults/defaultconfig.go

@@ -39,10 +39,6 @@ func DefaultConfig() *config.Config {
 			AssetPath:   filepath.Join(defaults.BaseDataPath(), "collaboration/fonts"),
 			PreviewText: "OpenCloud",
 		},
-		Events: config.Events{
-			Endpoint: "127.0.0.1:9233",
-			Cluster:  "opencloud-cluster",
-		},
 		Store: config.Store{
 			Store:    "nats-js-kv",
 			Nodes:    []string{"127.0.0.1:9233"},
@@ -96,10 +92,6 @@ func EnsureDefaults(cfg *config.Config) {
 	if cfg.CS3Api.GRPCClientTLS == nil && cfg.Commons != nil {
 		cfg.CS3Api.GRPCClientTLS = structs.CopyOrZeroValue(cfg.Commons.GRPCClientTLS)
 	}
-
-	if cfg.MachineAuthAPIKey == "" && cfg.Commons != nil && cfg.Commons.MachineAuthAPIKey != "" {
-		cfg.MachineAuthAPIKey = cfg.Commons.MachineAuthAPIKey
-	}
 }
 
 // Sanitize sanitized the configuration

services/collaboration/pkg/config/event.go

@@ -1,12 +0,0 @@
-package config
-
-// Events combines the configuration options for the event bus.
-type Events struct {
-	Endpoint             string `yaml:"endpoint" env:"OC_EVENTS_ENDPOINT;COLLABORATION_EVENTS_ENDPOINT" desc:"The address of the event system. The event system is the message queuing service. It is used as message broker for the microservice architecture." introductionVersion:"7.3.0"`
-	Cluster              string `yaml:"cluster" env:"OC_EVENTS_CLUSTER;COLLABORATION_EVENTS_CLUSTER" desc:"The clusterID of the event system. The event system is the message queuing service. It is used as message broker for the microservice architecture. Mandatory when using NATS as event system." introductionVersion:"7.3.0"`
-	TLSInsecure          bool   `yaml:"tls_insecure" env:"OC_INSECURE;OC_EVENTS_TLS_INSECURE;COLLABORATION_EVENTS_TLS_INSECURE" desc:"Whether to verify the server TLS certificates." introductionVersion:"7.3.0"`
-	TLSRootCACertificate string `yaml:"tls_root_ca_certificate" env:"OC_EVENTS_TLS_ROOT_CA_CERTIFICATE;COLLABORATION_EVENTS_TLS_ROOT_CA_CERTIFICATE" desc:"The root CA certificate used to validate the server's TLS certificate. If provided COLLABORATION_EVENTS_TLS_INSECURE will be seen as false." introductionVersion:"7.3.0"`
-	EnableTLS            bool   `yaml:"enable_tls" env:"OC_EVENTS_ENABLE_TLS;COLLABORATION_EVENTS_ENABLE_TLS" desc:"Enable TLS for the connection to the events broker. The events broker is the OpenCloud service which receives and delivers events between the services." introductionVersion:"7.3.0"`
-	AuthUsername         string `yaml:"username" env:"OC_EVENTS_AUTH_USERNAME;COLLABORATION_EVENTS_AUTH_USERNAME" desc:"The username to authenticate with the events broker. The events broker is the OpenCloud service which receives and delivers events between the services." introductionVersion:"7.3.0"`
-	AuthPassword         string `yaml:"password" env:"OC_EVENTS_AUTH_PASSWORD;COLLABORATION_EVENTS_AUTH_PASSWORD" desc:"The password to authenticate with the events broker. The events broker is the OpenCloud service which receives and delivers events between the services." introductionVersion:"7.3.0"`
-}

services/collaboration/pkg/notification/notification.go

@@ -1,10 +0,0 @@
-package notification
-
-import (
-	"github.com/go-playground/validator/v10"
-)
-
-var validate = validator.New(
-	validator.WithPrivateFieldValidation(),
-	validator.WithRequiredStructEnabled(),
-)

services/collaboration/pkg/notification/service.go

@@ -1,153 +0,0 @@
-package notification
-
-import (
-	"context"
-	"encoding/json"
-	"io"
-	"net/http"
-	"time"
-
-	gateway "github.com/cs3org/go-cs3apis/cs3/gateway/v1beta1"
-	rpcv1beta1 "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"
-	"github.com/opencloud-eu/reva/v2/pkg/events"
-	"github.com/opencloud-eu/reva/v2/pkg/rgrpc/todo/pool"
-	"github.com/opencloud-eu/reva/v2/pkg/storagespace"
-	"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/collaboration"
-)
-
-type ServiceOptions struct {
-	logger            log.Logger                                `validate:"required"`
-	eventPublisher    events.Publisher                          `validate:"required"`
-	gatewaySelector   pool.Selectable[gateway.GatewayAPIClient] `validate:"required"`
-	machineAuthAPIKey string                                    `validate:"required,min=1"`
-}
-
-func (o ServiceOptions) WithLogger(logger log.Logger) ServiceOptions {
-	o.logger = logger
-	return o
-}
-
-func (o ServiceOptions) WithEventPublisher(eventPublisher events.Publisher) ServiceOptions {
-	o.eventPublisher = eventPublisher
-	return o
-}
-
-func (o ServiceOptions) WithMachineAuthAPIKey(key string) ServiceOptions {
-	o.machineAuthAPIKey = key
-	return o
-}
-
-func (o ServiceOptions) WithGatewaySelector(gws pool.Selectable[gateway.GatewayAPIClient]) ServiceOptions {
-	o.gatewaySelector = gws
-	return o
-}
-
-type Service struct {
-	log               log.Logger
-	eventPublisher    events.Publisher
-	gatewaySelector   pool.Selectable[gateway.GatewayAPIClient]
-	machineAuthAPIKey string
-}
-
-func NewService(options ServiceOptions) (Service, error) {
-	if err := validate.Struct(options); err != nil {
-		return Service{}, err
-	}
-
-	return Service{
-		log:               options.logger,
-		eventPublisher:    options.eventPublisher,
-		gatewaySelector:   options.gatewaySelector,
-		machineAuthAPIKey: options.machineAuthAPIKey,
-	}, nil
-}
-
-func (s Service) HandleNotification(w http.ResponseWriter, r *http.Request) {
-	gatewayClient, err := s.gatewaySelector.Next()
-	if err != nil {
-		w.WriteHeader(http.StatusInternalServerError)
-		return
-	}
-
-	requestUser, canManage, err := collaboration.CheckPermissions(gatewayClient, r.Context(), collaboration.PermissionCollaborationPublishNotification)
-	switch {
-	case err != nil:
-		w.WriteHeader(http.StatusInternalServerError)
-		return
-	case !canManage:
-		w.WriteHeader(http.StatusForbidden)
-		return
-	}
-
-	defer func() { _ = r.Body.Close() }()
-	body, err := io.ReadAll(r.Body)
-	if err != nil {
-		w.WriteHeader(http.StatusInternalServerError)
-		return
-	}
-
-	var data = struct {
-		Type    string   `json:"type" validate:"required"`
-		UserIDs []string `json:"userIDs" validate:"required"`
-		FileID  string   `json:"fileID" validate:"required"`
-	}{}
-	if err := json.Unmarshal(body, &data); err != nil {
-		w.WriteHeader(http.StatusInternalServerError)
-		return
-	}
-
-	if err := validate.Struct(data); err != nil {
-		w.WriteHeader(http.StatusInternalServerError)
-		return
-	}
-
-	event := ocEvents.ResourceMention{
-		Executant: requestUser.GetId(),
-		Timestamp: time.Now(),
-	}
-
-	for _, userID := range data.UserIDs {
-		authResponse, err := gatewayClient.Authenticate(context.Background(), &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
-		}
-
-		statResponse, err := gatewayClient.Stat(
-			metadata.AppendToOutgoingContext(context.Background(), revactx.TokenHeader, authResponse.GetToken()),
-			&storageprovider.StatRequest{Ref: &storageprovider.Reference{ResourceId: &resourceID}},
-		)
-		if err != nil || statResponse.Status.Code != rpcv1beta1.Code_CODE_OK {
-			w.WriteHeader(http.StatusInternalServerError)
-			return
-		}
-
-		event.UserIDs = append(event.UserIDs, authResponse.User.GetId())
-		event.Ref = &storageprovider.Reference{
-			ResourceId: statResponse.GetInfo().GetId(),
-		}
-	}
-
-	if err := events.Publish(r.Context(), s.eventPublisher, event); err != nil {
-		w.WriteHeader(http.StatusInternalServerError)
-		return
-	}
-
-	w.WriteHeader(http.StatusOK)
-}

services/collaboration/pkg/server/http/option.go

@@ -10,7 +10,6 @@ import (
 	"github.com/opencloud-eu/opencloud/services/collaboration/pkg/config"
 	"github.com/opencloud-eu/opencloud/services/collaboration/pkg/connector"
 	"github.com/opencloud-eu/opencloud/services/collaboration/pkg/font"
-	"github.com/opencloud-eu/opencloud/services/collaboration/pkg/notification"
 )
 
 // Option defines a single option function.
@@ -18,14 +17,13 @@ type Option func(o *Options)
 
 // Options define the available options for this package.
 type Options struct {
-	Adapter             *connector.HttpAdapter
-	Logger              log.Logger
-	Context             context.Context
-	Config              *config.Config
-	TracerProvider      trace.TracerProvider
-	Store               microstore.Store
-	FontService         font.Service
-	NotificationService *notification.Service
+	Adapter        *connector.HttpAdapter
+	Logger         log.Logger
+	Context        context.Context
+	Config         *config.Config
+	TracerProvider trace.TracerProvider
+	Store          microstore.Store
+	FontService    font.Service
 }
 
 // newOptions initializes the available default options.
@@ -87,10 +85,3 @@ func FontService(val font.Service) Option {
 		o.FontService = val
 	}
 }
-
-// NotificationService provides a function to set the NotificationService option
-func NotificationService(val *notification.Service) Option {
-	return func(o *Options) {
-		o.NotificationService = val
-	}
-}

services/collaboration/pkg/server/http/server.go

@@ -22,10 +22,6 @@ import (
 func Server(opts ...Option) (http.Service, error) {
 	options := newOptions(opts...)
 
-	if options.NotificationService == nil {
-		options.Logger.Warn().Msg("running without notification service: no notifications will be sent, set the events endpoint to enable them")
-	}
-
 	service, err := http.NewService(
 		http.TLSConfig(options.Config.HTTP.TLS),
 		http.Logger(options.Logger),
@@ -100,7 +96,6 @@ func Server(opts ...Option) (http.Service, error) {
 // prepareRoutes will prepare all the implemented routes
 func prepareRoutes(r *chi.Mux, options Options) {
 	fontService := options.FontService
-	notificationService := options.NotificationService
 	adapter := options.Adapter
 	logger := options.Logger
 	// prepare basic logger for the request
@@ -232,11 +227,5 @@ func prepareRoutes(r *chi.Mux, options Options) {
 				r.Delete("/{id}", fontService.DeleteFont)
 			})
 		})
-
-		if notificationService != nil { // optional
-			r.With(auth).Route("/notify", func(r chi.Router) {
-				r.Post("/", notificationService.HandleNotification)
-			})
-		}
 	})
 }

services/graph/pkg/config/config.go

@@ -36,6 +36,8 @@ type Config struct {
 	Keycloak       Keycloak       `yaml:"keycloak"`
 	ServiceAccount ServiceAccount `yaml:"service_account"`
 
+	MachineAuthAPIKey string `yaml:"machine_auth_api_key" env:"OC_MACHINE_AUTH_API_KEY;GRAPH_MACHINE_AUTH_API_KEY" desc:"Machine auth API key used to validate internal requests necessary for the access to resources from other services." introductionVersion:"7.5.0" mask:"password"`
+
 	Context context.Context `yaml:"-"`
 
 	Metadata Metadata `yaml:"metadata_config"`

services/graph/pkg/config/defaults/defaultconfig.go

@@ -170,6 +170,10 @@ func EnsureDefaults(cfg *config.Config) {
 		cfg.HTTP.TLS = cfg.Commons.HTTPServiceTLS
 	}
 
+	if cfg.MachineAuthAPIKey == "" && cfg.Commons != nil && cfg.Commons.MachineAuthAPIKey != "" {
+		cfg.MachineAuthAPIKey = cfg.Commons.MachineAuthAPIKey
+	}
+
 	if cfg.Identity.LDAP.GroupCreateBaseDN == "" {
 		cfg.Identity.LDAP.GroupCreateBaseDN = cfg.Identity.LDAP.GroupBaseDN
 	}

services/graph/pkg/service/v0/service.go

@@ -341,6 +341,7 @@ func NewService(opts ...Option) (Graph, error) { //nolint:maintidx
 					r.Get("/", svc.GetUser)
 					r.Get("/drive", svc.GetUserDrive)
 					r.Post("/exportPersonalData", svc.ExportPersonalData)
+					r.Post("/teamwork/sendActivityNotification", svc.SendActivityNotification)
 					r.Route("/photo/$value", func(r chi.Router) {
 						r.Get("/", usersUserProfilePhotoApi.GetProfilePhoto(GetSlugValue("userID")))
 					})

services/graph/pkg/service/v0/teamwork.go

@@ -0,0 +1,197 @@
+package svc
+
+import (
+	"context"
+	"net/http"
+	"net/url"
+	"time"
+
+	gateway "github.com/cs3org/go-cs3apis/cs3/gateway/v1beta1"
+	userpb "github.com/cs3org/go-cs3apis/cs3/identity/user/v1beta1"
+	rpc "github.com/cs3org/go-cs3apis/cs3/rpc/v1beta1"
+	storageprovider "github.com/cs3org/go-cs3apis/cs3/storage/provider/v1beta1"
+	"github.com/go-chi/chi/v5"
+	revactx "github.com/opencloud-eu/reva/v2/pkg/ctx"
+	"github.com/opencloud-eu/reva/v2/pkg/events"
+	"github.com/opencloud-eu/reva/v2/pkg/storagespace"
+	"google.golang.org/grpc/metadata"
+
+	ocEvents "github.com/opencloud-eu/opencloud/pkg/events"
+	settingssvc "github.com/opencloud-eu/opencloud/protogen/gen/opencloud/services/settings/v0"
+	"github.com/opencloud-eu/opencloud/services/graph/pkg/errorcode"
+	settingsServiceExt "github.com/opencloud-eu/opencloud/services/settings/pkg/store/defaults"
+)
+
+const (
+	// WebOfficeAppID identifies the app a notification comes from
+	WebOfficeAppID = "8d1c9c88-9e2c-4d0b-9a1e-6a9de1cb9d3c"
+
+	_activityTypeMentioned = "mentioned"
+	_topicSourceText       = "text"
+)
+
+type activityTopic struct {
+	Source string `json:"source"`
+	Value  string `json:"value"`
+}
+
+type activityNotification struct {
+	Topic        activityTopic `json:"topic"`
+	ActivityType string        `json:"activityType"`
+	TeamsAppID   string        `json:"teamsAppId"`
+}
+
+// SendActivityNotification tells a user that they were named on a resource.
+func (g Graph) SendActivityNotification(w http.ResponseWriter, r *http.Request) {
+	ctx := r.Context()
+
+	if g.eventsPublisher == nil {
+		g.logger.Error().Msg("no events publisher configured, activity notifications cannot be delivered")
+		errorcode.ServiceNotAvailable.Render(w, r, http.StatusServiceUnavailable, "activity notifications are not available")
+		return
+	}
+
+	executant, ok := revactx.ContextGetUser(ctx)
+	if !ok {
+		errorcode.GeneralException.Render(w, r, http.StatusUnauthorized, "user not found in context")
+		return
+	}
+
+	userID, err := url.PathUnescape(chi.URLParam(r, "userID"))
+	if err != nil || userID == "" {
+		errorcode.InvalidRequest.Render(w, r, http.StatusBadRequest, "invalid user id")
+		return
+	}
+
+	notification := activityNotification{}
+	if err := StrictJSONUnmarshal(r.Body, &notification); err != nil {
+		g.logger.Debug().Err(err).Msg("could not parse the activity notification")
+		errorcode.InvalidRequest.Render(w, r, http.StatusBadRequest, "invalid request body")
+		return
+	}
+
+	switch {
+	case notification.ActivityType != _activityTypeMentioned:
+		errorcode.InvalidRequest.Render(w, r, http.StatusBadRequest, "unsupported activityType")
+		return
+	case notification.TeamsAppID != WebOfficeAppID:
+		errorcode.InvalidRequest.Render(w, r, http.StatusBadRequest, "unknown teamsAppId")
+		return
+	case notification.Topic.Source != _topicSourceText:
+		errorcode.InvalidRequest.Render(w, r, http.StatusBadRequest, "unsupported topic source")
+		return
+	}
+
+	itemID, err := storagespace.ParseID(notification.Topic.Value)
+	if err != nil || itemID.GetStorageId() == "" || itemID.GetSpaceId() == "" || itemID.GetOpaqueId() == "" {
+		g.logger.Debug().Err(err).Msg("could not parse the topic value")
+		errorcode.InvalidRequest.Render(w, r, http.StatusBadRequest, "the topic value is no resource id")
+		return
+	}
+
+	if _, err := g.permissionsService.GetPermissionByID(ctx, &settingssvc.GetPermissionByIDRequest{
+		PermissionId: settingsServiceExt.CollaborationPublishNotificationPermission(0).Id,
+	}); err != nil {
+		g.logger.Debug().Err(err).Msg("user is not allowed to publish activity notifications")
+		errorcode.AccessDenied.Render(w, r, http.StatusForbidden, "not allowed to publish activity notifications")
+		return
+	}
+
+	gatewayClient, err := g.gatewaySelector.Next()
+	if err != nil {
+		g.logger.Error().Err(err).Msg("could not select next gateway client")
+		errorcode.ServiceNotAvailable.Render(w, r, http.StatusServiceUnavailable, "could not select next gateway client")
+		return
+	}
+
+	ref := &storageprovider.Reference{ResourceId: &itemID}
+
+	statResponse, err := gatewayClient.Stat(ctx, &storageprovider.StatRequest{Ref: ref})
+	switch {
+	case err != nil:
+		g.logger.Error().Err(err).Msg("could not stat item")
+		errorcode.GeneralException.Render(w, r, http.StatusInternalServerError, "could not stat item")
+		return
+	case statResponse.GetStatus().GetCode() == rpc.Code_CODE_NOT_FOUND,
+		statResponse.GetStatus().GetCode() == rpc.Code_CODE_PERMISSION_DENIED:
+		errorcode.ItemNotFound.Render(w, r, http.StatusNotFound, "item not found")
+		return
+	case statResponse.GetStatus().GetCode() != rpc.Code_CODE_OK:
+		g.logger.Error().
+			Str("code", statResponse.GetStatus().GetCode().String()).
+			Str("message", statResponse.GetStatus().GetMessage()).
+			Msg("could not stat item")
+		errorcode.RenderError(w, r, errorcode.FromCS3Status(statResponse.GetStatus(), nil))
+		return
+	}
+
+	authResponse, err := gatewayClient.Authenticate(ctx, &gateway.AuthenticateRequest{
+		Type:         "machine",
+		ClientId:     "userid:" + userID,
+		ClientSecret: g.config.MachineAuthAPIKey,
+	})
+	switch {
+	case err != nil:
+		g.logger.Error().Err(err).Msg("could not authenticate the recipient")
+		errorcode.GeneralException.Render(w, r, http.StatusInternalServerError, "could not authenticate the recipient")
+		return
+	case authResponse.GetStatus().GetCode() == rpc.Code_CODE_NOT_FOUND:
+		g.logger.Debug().Str("userID", userID).Msg("the recipient does not exist")
+		errorcode.ItemNotFound.Render(w, r, http.StatusNotFound, "recipient not found")
+		return
+	case authResponse.GetStatus().GetCode() != rpc.Code_CODE_OK:
+		g.logger.Error().
+			Str("userID", userID).
+			Str("code", authResponse.GetStatus().GetCode().String()).
+			Str("message", authResponse.GetStatus().GetMessage()).
+			Msg("could not authenticate the recipient")
+		errorcode.RenderError(w, r, errorcode.FromCS3Status(authResponse.GetStatus(), nil))
+		return
+	}
+
+	recipientStat, err := gatewayClient.Stat(
+		func() context.Context {
+			md, _ := metadata.FromOutgoingContext(ctx)
+			md = md.Copy()
+			md.Set(revactx.TokenHeader, authResponse.GetToken())
+
+			return metadata.NewOutgoingContext(ctx, md)
+		}(),
+		&storageprovider.StatRequest{Ref: ref},
+	)
+
+	switch {
+	case err != nil:
+		g.logger.Error().Err(err).Msg("could not stat the item as the recipient")
+		errorcode.GeneralException.Render(w, r, http.StatusInternalServerError, "could not stat item")
+		return
+	case recipientStat.GetStatus().GetCode() == rpc.Code_CODE_NOT_FOUND,
+		recipientStat.GetStatus().GetCode() == rpc.Code_CODE_PERMISSION_DENIED:
+		g.logger.Debug().Str("userID", userID).Msg("mention dropped, the recipient has no access to the item")
+		w.WriteHeader(http.StatusAccepted)
+		return
+	case recipientStat.GetStatus().GetCode() != rpc.Code_CODE_OK:
+		g.logger.Error().
+			Str("userID", userID).
+			Str("code", recipientStat.GetStatus().GetCode().String()).
+			Str("message", recipientStat.GetStatus().GetMessage()).
+			Msg("could not stat the item as the recipient")
+		errorcode.RenderError(w, r, errorcode.FromCS3Status(recipientStat.GetStatus(), nil))
+		return
+	}
+
+	event := ocEvents.ResourceMention{
+		Executant: executant.GetId(),
+		UserIDs:   []*userpb.UserId{authResponse.GetUser().GetId()},
+		Ref:       &storageprovider.Reference{ResourceId: statResponse.GetInfo().GetId()},
+		Timestamp: time.Now(),
+	}
+
+	if err := events.Publish(ctx, g.eventsPublisher, event); err != nil {
+		g.logger.Error().Err(err).Msg("could not publish the activity notification")
+		errorcode.GeneralException.Render(w, r, http.StatusInternalServerError, "could not publish the activity notification")
+		return
+	}
+
+	w.WriteHeader(http.StatusAccepted)
+}

services/graph/pkg/service/v0/teamwork_test.go

@@ -0,0 +1,362 @@
+package svc_test
+
+import (
+	"context"
+	"errors"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+
+	gateway "github.com/cs3org/go-cs3apis/cs3/gateway/v1beta1"
+	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/go-chi/chi/v5"
+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
+	revactx "github.com/opencloud-eu/reva/v2/pkg/ctx"
+	"github.com/opencloud-eu/reva/v2/pkg/rgrpc/status"
+	"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"
+	"google.golang.org/grpc"
+	grpcmetadata "google.golang.org/grpc/metadata"
+
+	ocEvents "github.com/opencloud-eu/opencloud/pkg/events"
+	"github.com/opencloud-eu/opencloud/pkg/shared"
+	settings "github.com/opencloud-eu/opencloud/protogen/gen/opencloud/services/settings/v0"
+	"github.com/opencloud-eu/opencloud/services/graph/mocks"
+	"github.com/opencloud-eu/opencloud/services/graph/pkg/config"
+	"github.com/opencloud-eu/opencloud/services/graph/pkg/config/defaults"
+	service "github.com/opencloud-eu/opencloud/services/graph/pkg/service/v0"
+)
+
+var _ = Describe("SendActivityNotification", func() {
+	var (
+		svc               service.Graph
+		cfg               *config.Config
+		gatewayClient     *cs3mocks.GatewayAPIClient
+		eventsPublisher   mocks.Publisher
+		permissionService *mocks.Permissions
+
+		currentUser = &userpb.User{Id: &userpb.UserId{OpaqueId: "executant"}}
+
+		mention = `{"topic":{"source":"text","value":"storage$space!item"},` +
+			`"activityType":"mentioned","teamsAppId":"` + service.WebOfficeAppID + `"}`
+	)
+
+	// the recipient is the user in the path, so every request names one
+	request := func(userID, body string) *http.Request {
+		ctx := revactx.ContextSetUser(context.Background(), currentUser)
+
+		rctx := chi.NewRouteContext()
+		rctx.URLParams.Add("userID", userID)
+
+		return httptest.NewRequest(
+			http.MethodPost,
+			"/graph/v1.0/users/"+userID+"/teamwork/sendActivityNotification",
+			strings.NewReader(body),
+		).WithContext(context.WithValue(ctx, chi.RouteCtxKey, rctx))
+	}
+
+	mentions := func() []ocEvents.ResourceMention {
+		var mentions []ocEvents.ResourceMention
+		for _, call := range eventsPublisher.Calls {
+			if mention, ok := call.Arguments[1].(ocEvents.ResourceMention); ok {
+				mentions = append(mentions, mention)
+			}
+		}
+
+		return mentions
+	}
+
+	// the stat answers with the status the callback picks for the token the call carries
+	statWith := func(statusFor func(ctx context.Context, token string) *rpc.Status) {
+		gatewayClient.On("Stat", mock.Anything, mock.Anything).Return(
+			func(ctx context.Context, _ *provider.StatRequest, _ ...grpc.CallOption) *provider.StatResponse {
+				var token string
+				if md, ok := grpcmetadata.FromOutgoingContext(ctx); ok {
+					token = strings.Join(md.Get(revactx.TokenHeader), "")
+				}
+
+				st := statusFor(ctx, token)
+				if st.GetCode() != rpc.Code_CODE_OK {
+					return &provider.StatResponse{Status: st}
+				}
+
+				return &provider.StatResponse{
+					Status: st,
+					Info: &provider.ResourceInfo{
+						Id: &provider.ResourceId{StorageId: "storage", SpaceId: "space", OpaqueId: "item"},
+					},
+				}
+			}, nil)
+	}
+
+	statAs := func(tokens ...string) {
+		allowed := make(map[string]struct{}, len(tokens))
+		for _, token := range tokens {
+			allowed[token] = struct{}{}
+		}
+
+		statWith(func(ctx context.Context, token string) *rpc.Status {
+			if _, ok := allowed[token]; !ok {
+				return status.NewNotFound(ctx, "not found")
+			}
+
+			return status.NewOK(ctx)
+		})
+	}
+
+	BeforeEach(func() {
+		eventsPublisher = mocks.Publisher{}
+		eventsPublisher.On("Publish", mock.Anything, mock.Anything, mock.Anything).Return(nil)
+
+		permissionService = &mocks.Permissions{}
+		permissionService.On("GetPermissionByID", mock.Anything, mock.Anything).
+			Return(&settings.GetPermissionByIDResponse{}, nil)
+
+		pool.RemoveSelector("GatewaySelector" + "eu.opencloud.api.gateway")
+		gatewayClient = &cs3mocks.GatewayAPIClient{}
+		gatewaySelector := pool.GetSelector[gateway.GatewayAPIClient](
+			"GatewaySelector",
+			"eu.opencloud.api.gateway",
+			func(cc grpc.ClientConnInterface) gateway.GatewayAPIClient {
+				return gatewayClient
+			},
+		)
+
+		gatewayClient.On("Authenticate", mock.Anything, mock.Anything).Return(
+			func(_ context.Context, req *gateway.AuthenticateRequest, _ ...grpc.CallOption) *gateway.AuthenticateResponse {
+				userID := strings.TrimPrefix(req.GetClientId(), "userid:")
+				switch userID {
+				case "nobody":
+					return &gateway.AuthenticateResponse{Status: status.NewNotFound(context.Background(), "not found")}
+				case "broken":
+					return &gateway.AuthenticateResponse{Status: status.NewInternal(context.Background(), "auth failed")}
+				case "denied":
+					return &gateway.AuthenticateResponse{Status: status.NewPermissionDenied(context.Background(), nil, "permission denied")}
+				}
+
+				return &gateway.AuthenticateResponse{
+					Status: status.NewOK(context.Background()),
+					Token:  userID + "-token",
+					User:   &userpb.User{Id: &userpb.UserId{OpaqueId: userID}},
+				}
+			}, nil)
+
+		cfg = defaults.FullDefaultConfig()
+		cfg.Identity.LDAP.CACert = ""
+		cfg.TokenManager.JWTSecret = "loremipsum"
+		cfg.Commons = &shared.Commons{}
+		cfg.GRPCClientTLS = &shared.GRPCClientTLS{}
+		cfg.MachineAuthAPIKey = "machine-auth-api-key"
+
+		var err error
+		svc, err = service.NewService(
+			service.Config(cfg),
+			service.WithGatewaySelector(gatewaySelector),
+			service.EventsPublisher(&eventsPublisher),
+			service.PermissionService(permissionService),
+		)
+		Expect(err).ToNot(HaveOccurred())
+	})
+
+	It("notifies the user from the path", func() {
+		statAs("", "alice-token")
+
+		rr := httptest.NewRecorder()
+		svc.SendActivityNotification(rr, request("alice", mention))
+
+		Expect(rr.Code).To(Equal(http.StatusAccepted))
+		Expect(mentions()).To(HaveLen(1))
+		Expect(mentions()[0].Executant.GetOpaqueId()).To(Equal("executant"))
+		Expect(mentions()[0].UserIDs).To(HaveLen(1))
+		Expect(mentions()[0].UserIDs[0].GetOpaqueId()).To(Equal("alice"))
+		Expect(mentions()[0].Ref.GetResourceId().GetOpaqueId()).To(Equal("item"))
+	})
+
+	It("is unavailable without an events publisher", func() {
+		var err error
+		svc, err = service.NewService(
+			service.Config(cfg),
+			service.WithGatewaySelector(pool.GetSelector[gateway.GatewayAPIClient](
+				"GatewaySelector",
+				"eu.opencloud.api.gateway",
+				func(cc grpc.ClientConnInterface) gateway.GatewayAPIClient {
+					return gatewayClient
+				},
+			)),
+			service.PermissionService(permissionService),
+		)
+		Expect(err).ToNot(HaveOccurred())
+		statAs("", "alice-token")
+
+		rr := httptest.NewRecorder()
+		svc.SendActivityNotification(rr, request("alice", mention))
+
+		Expect(rr.Code).To(Equal(http.StatusServiceUnavailable))
+		Expect(mentions()).To(BeEmpty())
+	})
+
+	It("denies a caller without the publish permission", func() {
+		permissionService.ExpectedCalls = nil
+		permissionService.On("GetPermissionByID", mock.Anything, mock.Anything).
+			Return(nil, errors.New("not found"))
+		statAs("", "alice-token")
+
+		rr := httptest.NewRecorder()
+		svc.SendActivityNotification(rr, request("alice", mention))
+
+		Expect(rr.Code).To(Equal(http.StatusForbidden))
+		Expect(mentions()).To(BeEmpty())
+	})
+
+	It("hides the item from a caller who cannot see it", func() {
+		statAs("alice-token")
+
+		rr := httptest.NewRecorder()
+		svc.SendActivityNotification(rr, request("alice", mention))
+
+		Expect(rr.Code).To(Equal(http.StatusNotFound))
+		Expect(mentions()).To(BeEmpty())
+	})
+
+	It("fails when the item cannot be stated as the caller", func() {
+		statWith(func(ctx context.Context, _ string) *rpc.Status {
+			return status.NewInternal(ctx, "stat failed")
+		})
+
+		rr := httptest.NewRecorder()
+		svc.SendActivityNotification(rr, request("alice", mention))
+
+		Expect(rr.Code).To(Equal(http.StatusInternalServerError))
+		Expect(mentions()).To(BeEmpty())
+	})
+
+	// anything the caller stat answers beyond that is whatever the cs3 status maps to
+	It("carries the cs3 status of a failed caller stat", func() {
+		statWith(func(ctx context.Context, _ string) *rpc.Status {
+			return status.NewInvalidArg(ctx, "invalid reference")
+		})
+
+		rr := httptest.NewRecorder()
+		svc.SendActivityNotification(rr, request("alice", mention))
+
+		Expect(rr.Code).To(Equal(http.StatusBadRequest))
+		Expect(mentions()).To(BeEmpty())
+	})
+
+	// a recipient without access looks like success, so the sender cannot probe who has it
+	It("silently drops a mention for a recipient who cannot see the item", func() {
+		statAs("")
+
+		rr := httptest.NewRecorder()
+		svc.SendActivityNotification(rr, request("alice", mention))
+
+		Expect(rr.Code).To(Equal(http.StatusAccepted))
+		Expect(mentions()).To(BeEmpty())
+	})
+
+	It("silently drops a mention for a recipient who may not see the item", func() {
+		statWith(func(ctx context.Context, token string) *rpc.Status {
+			if token == "alice-token" {
+				return status.NewPermissionDenied(ctx, nil, "permission denied")
+			}
+
+			return status.NewOK(ctx)
+		})
+
+		rr := httptest.NewRecorder()
+		svc.SendActivityNotification(rr, request("alice", mention))
+
+		Expect(rr.Code).To(Equal(http.StatusAccepted))
+		Expect(mentions()).To(BeEmpty())
+	})
+
+	// only a not found and a permission denied are the recipient's own, the rest is ours
+	It("fails when the item cannot be stated as the recipient", func() {
+		statWith(func(ctx context.Context, token string) *rpc.Status {
+			if token == "alice-token" {
+				return status.NewInternal(ctx, "stat failed")
+			}
+
+			return status.NewOK(ctx)
+		})
+
+		rr := httptest.NewRecorder()
+		svc.SendActivityNotification(rr, request("alice", mention))
+
+		Expect(rr.Code).To(Equal(http.StatusInternalServerError))
+		Expect(mentions()).To(BeEmpty())
+	})
+
+	It("carries the cs3 status of a failed recipient stat", func() {
+		statWith(func(ctx context.Context, token string) *rpc.Status {
+			if token == "alice-token" {
+				return status.NewLocked(ctx, "locked")
+			}
+
+			return status.NewOK(ctx)
+		})
+
+		rr := httptest.NewRecorder()
+		svc.SendActivityNotification(rr, request("alice", mention))
+
+		Expect(rr.Code).To(Equal(http.StatusLocked))
+		Expect(mentions()).To(BeEmpty())
+	})
+
+	// a user id is no secret, other endpoints look users up as well
+	It("refuses a recipient that does not exist", func() {
+		statAs("", "alice-token")
+
+		rr := httptest.NewRecorder()
+		svc.SendActivityNotification(rr, request("nobody", mention))
+
+		Expect(rr.Code).To(Equal(http.StatusNotFound))
+		Expect(mentions()).To(BeEmpty())
+	})
+
+	It("fails when the recipient cannot be authenticated", func() {
+		statAs("")
+
+		rr := httptest.NewRecorder()
+		svc.SendActivityNotification(rr, request("broken", mention))
+
+		Expect(rr.Code).To(Equal(http.StatusInternalServerError))
+		Expect(mentions()).To(BeEmpty())
+	})
+
+	// a rejected machine auth is a server side misconfiguration, the status decides what it looks like
+	It("carries the cs3 status of a rejected machine auth", func() {
+		statAs("")
+
+		rr := httptest.NewRecorder()
+		svc.SendActivityNotification(rr, request("denied", mention))
+
+		Expect(rr.Code).To(Equal(http.StatusForbidden))
+		Expect(mentions()).To(BeEmpty())
+	})
+
+	DescribeTable("rejects a malformed body",
+		func(body string) {
+			statAs("", "alice-token")
+
+			rr := httptest.NewRecorder()
+			svc.SendActivityNotification(rr, request("alice", body))
+
+			Expect(rr.Code).To(Equal(http.StatusBadRequest))
+			Expect(mentions()).To(BeEmpty())
+		},
+		Entry("no json", `not json`),
+		Entry("unknown field", `{"topic":{"source":"text","value":"storage$space!item"},"activityType":"mentioned","teamsAppId":"`+service.WebOfficeAppID+`","chainId":1}`),
+		Entry("template parameters", `{"topic":{"source":"text","value":"storage$space!item"},"activityType":"mentioned","teamsAppId":"`+service.WebOfficeAppID+`","templateParameters":[{"name":"actor","value":"someone else"}]}`),
+		Entry("no topic", `{"activityType":"mentioned","teamsAppId":"`+service.WebOfficeAppID+`"}`),
+		Entry("topic source entityUrl", `{"topic":{"source":"entityUrl","value":"https://cloud.opencloud.test/f/item"},"activityType":"mentioned","teamsAppId":"`+service.WebOfficeAppID+`"}`),
+		Entry("topic value is no resource id", `{"topic":{"source":"text","value":"not-an-id"},"activityType":"mentioned","teamsAppId":"`+service.WebOfficeAppID+`"}`),
+		Entry("unknown activityType", `{"topic":{"source":"text","value":"storage$space!item"},"activityType":"reactedTo","teamsAppId":"`+service.WebOfficeAppID+`"}`),
+		Entry("unknown teamsAppId", `{"topic":{"source":"text","value":"storage$space!item"},"activityType":"mentioned","teamsAppId":"14a4bd3a-1e0f-4a2e-9f30-1cc1f0d0a1cd"}`),
+		Entry("no teamsAppId", `{"topic":{"source":"text","value":"storage$space!item"},"activityType":"mentioned"}`),
+	)
+})

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")

services/proxy/pkg/config/defaults/defaultconfig.go

@@ -310,11 +310,6 @@ func DefaultPolicies() []config.Policy {
 					Service:  "eu.opencloud.web.collaboration",
 					// Method: "POST" // toDo: fails with method, WHY???
 				},
-				{
-					Endpoint: "/collaboration/notify",
-					Service:  "eu.opencloud.web.collaboration",
-					// Method: "POST" // toDo: fails with method, WHY???
-				},
 				{
 					Endpoint:    "/collaboration",
 					Service:     "eu.opencloud.web.collaboration",