implement user and group audit events

graph/Makefile

@@ -27,6 +27,8 @@ include ../.make/generate.mk
 ci-go-generate: $(MOCKERY) # CI runs ci-node-generate automatically before this target
 	$(MOCKERY) --dir pkg/service/v0 --case underscore --name GatewayClient
 	$(MOCKERY) --dir pkg/service/v0 --case underscore --name HTTPClient
+	$(MOCKERY) --dir pkg/service/v0 --case underscore --name Publisher
+
 
 .PHONY: ci-node-generate
 ci-node-generate:

graph/mocks/publisher.go

@@ -0,0 +1,34 @@
+// Code generated by mockery v2.9.4. DO NOT EDIT.
+
+package mocks
+
+import (
+	mock "github.com/stretchr/testify/mock"
+	events "go-micro.dev/v4/events"
+)
+
+// Publisher is an autogenerated mock type for the Publisher type
+type Publisher struct {
+	mock.Mock
+}
+
+// Publish provides a mock function with given fields: _a0, _a1, _a2
+func (_m *Publisher) Publish(_a0 string, _a1 interface{}, _a2 ...events.PublishOption) error {
+	_va := make([]interface{}, len(_a2))
+	for _i := range _a2 {
+		_va[_i] = _a2[_i]
+	}
+	var _ca []interface{}
+	_ca = append(_ca, _a0, _a1)
+	_ca = append(_ca, _va...)
+	ret := _m.Called(_ca...)
+
+	var r0 error
+	if rf, ok := ret.Get(0).(func(string, interface{}, ...events.PublishOption) error); ok {
+		r0 = rf(_a0, _a1, _a2...)
+	} else {
+		r0 = ret.Error(0)
+	}
+
+	return r0
+}

graph/pkg/config/config.go

@@ -23,6 +23,7 @@ type Config struct {
 
 	Spaces   Spaces   `yaml:"spaces"`
 	Identity Identity `yaml:"identity"`
+	Events   Events   `yaml:"events"`
 
 	Context context.Context `yaml:"-"`
 }
@@ -62,3 +63,9 @@ type Identity struct {
 	Backend string `yaml:"backend" env:"GRAPH_IDENTITY_BACKEND"`
 	LDAP    LDAP   `yaml:"ldap"`
 }
+
+// Events combines the configuration options for the event bus.
+type Events struct {
+	Endpoint string `yaml:"events_endpoint" env:"GRAPH_EVENTS_ENDPOINT" desc:"the address of the streaming service"`
+	Cluster  string `yaml:"events_cluster" env:"GRAPH_EVENTS_CLUSTER" desc:"the clusterID of the streaming service. Mandatory when using nats"`
+}

graph/pkg/config/defaults/defaultconfig.go

@@ -57,6 +57,10 @@ func DefaultConfig() *config.Config {
 				GroupIDAttribute:   "owncloudUUID",
 			},
 		},
+		Events: config.Events{
+			Endpoint: "127.0.0.1:9233",
+			Cluster:  "ocis-cluster",
+		},
 	}
 }
 

graph/pkg/server/http/server.go

@@ -1,6 +1,8 @@
 package http
 
 import (
+	"github.com/asim/go-micro/plugins/events/natsjs/v4"
+	"github.com/cs3org/reva/v2/pkg/events/server"
 	chimiddleware "github.com/go-chi/chi/v5/middleware"
 	graphMiddleware "github.com/owncloud/ocis/graph/pkg/middleware"
 	svc "github.com/owncloud/ocis/graph/pkg/service/v0"
@@ -8,6 +10,7 @@ import (
 	"github.com/owncloud/ocis/ocis-pkg/middleware"
 	"github.com/owncloud/ocis/ocis-pkg/service/http"
 	"github.com/owncloud/ocis/ocis-pkg/version"
+	"github.com/pkg/errors"
 	"go-micro.dev/v4"
 )
 
@@ -25,6 +28,17 @@ func Server(opts ...Option) (http.Service, error) {
 		http.Flags(options.Flags...),
 	)
 
+	publisher, err := server.NewNatsStream(
+		natsjs.Address(options.Config.Events.Endpoint),
+		natsjs.ClusterID(options.Config.Events.Cluster),
+	)
+	if err != nil {
+		options.Logger.Error().
+			Err(err).
+			Msg("Error initializing events publisher")
+		return http.Service{}, errors.Wrap(err, "could not initialize events publisher")
+	}
+
 	handle := svc.NewService(
 		svc.Logger(options.Logger),
 		svc.Config(options.Config),
@@ -42,6 +56,7 @@ func Server(opts ...Option) (http.Service, error) {
 				account.JWTSecret(options.Config.TokenManager.JWTSecret),
 			),
 		),
+		svc.EventsPublisher(publisher),
 	)
 
 	{

graph/pkg/service/v0/graph.go

@@ -7,11 +7,13 @@ import (
 	"github.com/ReneKroon/ttlcache/v2"
 	gateway "github.com/cs3org/go-cs3apis/cs3/gateway/v1beta1"
 	provider "github.com/cs3org/go-cs3apis/cs3/storage/provider/v1beta1"
+	"github.com/cs3org/reva/v2/pkg/events"
 	"github.com/go-chi/chi/v5"
 	"github.com/owncloud/ocis/graph/pkg/config"
 	"github.com/owncloud/ocis/graph/pkg/identity"
 	"github.com/owncloud/ocis/ocis-pkg/log"
 	settingssvc "github.com/owncloud/ocis/protogen/gen/ocis/services/settings/v0"
+	mevents "go-micro.dev/v4/events"
 	"google.golang.org/grpc"
 )
 
@@ -51,6 +53,11 @@ type GatewayClient interface {
 	GetQuota(ctx context.Context, in *gateway.GetQuotaRequest, opts ...grpc.CallOption) (*provider.GetQuotaResponse, error)
 }
 
+// Publisher is the interface for events publisher
+type Publisher interface {
+	Publish(string, interface{}, ...mevents.PublishOption) error
+}
+
 // HTTPClient is the subset of the http.Client that is being used to interact with the download gateway
 type HTTPClient interface {
 	Do(req *http.Request) (*http.Response, error)
@@ -69,6 +76,7 @@ type Graph struct {
 	httpClient           HTTPClient
 	roleService          settingssvc.RoleService
 	spacePropertiesCache *ttlcache.Cache
+	eventsPublisher      events.Publisher
 }
 
 // ServeHTTP implements the Service interface.
@@ -86,6 +94,14 @@ func (g Graph) GetHTTPClient() HTTPClient {
 	return g.httpClient
 }
 
+func (g Graph) publishEvent(ev interface{}) {
+	if err := events.Publish(g.eventsPublisher, ev); err != nil {
+		g.logger.Error().
+			Err(err).
+			Msg("could not publish user created event")
+	}
+}
+
 type listResponse struct {
 	Value interface{} `json:"value,omitempty"`
 }

graph/pkg/service/v0/graph_test.go

@@ -25,20 +25,23 @@ import (
 
 var _ = Describe("Graph", func() {
 	var (
-		svc           service.Service
-		gatewayClient *mocks.GatewayClient
-		httpClient    *mocks.HTTPClient
-		ctx           context.Context
+		svc             service.Service
+		gatewayClient   *mocks.GatewayClient
+		httpClient      *mocks.HTTPClient
+		eventsPublisher mocks.Publisher
+		ctx             context.Context
 	)
 
 	JustBeforeEach(func() {
 		ctx = context.Background()
 		gatewayClient = &mocks.GatewayClient{}
 		httpClient = &mocks.HTTPClient{}
+		eventsPublisher = mocks.Publisher{}
 		svc = service.NewService(
 			service.Config(defaults.DefaultConfig()),
 			service.WithGatewayClient(gatewayClient),
 			service.WithHTTPClient(httpClient),
+			service.EventsPublisher(&eventsPublisher),
 		)
 	})
 

graph/pkg/service/v0/groups.go

@@ -13,6 +13,7 @@ import (
 	libregraph "github.com/owncloud/libre-graph-api-go"
 	"github.com/owncloud/ocis/graph/pkg/service/v0/errorcode"
 
+	"github.com/cs3org/reva/v2/pkg/events"
 	"github.com/go-chi/chi/v5"
 	"github.com/go-chi/render"
 )
@@ -81,6 +82,9 @@ func (g Graph) PostGroup(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
+	if grp != nil && grp.Id != nil {
+		g.publishEvent(events.GroupCreated{GroupID: *grp.Id})
+	}
 	render.Status(r, http.StatusOK)
 	render.JSON(w, r, grp)
 }
@@ -197,6 +201,8 @@ func (g Graph) DeleteGroup(w http.ResponseWriter, r *http.Request) {
 		}
 		return
 	}
+
+	g.publishEvent(events.GroupDeleted{GroupID: groupID})
 	render.Status(r, http.StatusNoContent)
 	render.NoContent(w, r)
 }
@@ -279,6 +285,8 @@ func (g Graph) PostGroupMember(w http.ResponseWriter, r *http.Request) {
 		}
 		return
 	}
+
+	g.publishEvent(events.GroupMemberAdded{GroupID: groupID, UserID: id})
 	render.Status(r, http.StatusNoContent)
 	render.NoContent(w, r)
 }
@@ -322,6 +330,7 @@ func (g Graph) DeleteGroupMember(w http.ResponseWriter, r *http.Request) {
 		}
 		return
 	}
+	g.publishEvent(events.GroupMemberRemoved{GroupID: groupID, UserID: memberID})
 	render.Status(r, http.StatusNoContent)
 	render.NoContent(w, r)
 }

graph/pkg/service/v0/option.go

@@ -3,6 +3,7 @@ package svc
 import (
 	"net/http"
 
+	"github.com/cs3org/reva/v2/pkg/events"
 	"github.com/owncloud/ocis/graph/pkg/config"
 	"github.com/owncloud/ocis/ocis-pkg/log"
 	"github.com/owncloud/ocis/ocis-pkg/roles"
@@ -14,13 +15,14 @@ type Option func(o *Options)
 
 // Options defines the available options for this package.
 type Options struct {
-	Logger        log.Logger
-	Config        *config.Config
-	Middleware    []func(http.Handler) http.Handler
-	GatewayClient GatewayClient
-	HTTPClient    HTTPClient
-	RoleService   settingssvc.RoleService
-	RoleManager   *roles.Manager
+	Logger          log.Logger
+	Config          *config.Config
+	Middleware      []func(http.Handler) http.Handler
+	GatewayClient   GatewayClient
+	HTTPClient      HTTPClient
+	RoleService     settingssvc.RoleService
+	RoleManager     *roles.Manager
+	EventsPublisher events.Publisher
 }
 
 // newOptions initializes the available default options.
@@ -82,3 +84,10 @@ func RoleManager(val *roles.Manager) Option {
 		o.RoleManager = val
 	}
 }
+
+// EventsPublisher provides a function to set the EventsPublisher option.
+func EventsPublisher(val events.Publisher) Option {
+	return func(o *Options) {
+		o.EventsPublisher = val
+	}
+}

graph/pkg/service/v0/service.go

@@ -96,6 +96,7 @@ func NewService(opts ...Option) Service {
 		logger:               &options.Logger,
 		identityBackend:      backend,
 		spacePropertiesCache: ttlcache.NewCache(),
+		eventsPublisher:      options.EventsPublisher,
 	}
 	if options.GatewayClient == nil {
 		var err error

graph/pkg/service/v0/users.go

@@ -12,6 +12,7 @@ import (
 
 	"github.com/CiscoM31/godata"
 	revactx "github.com/cs3org/reva/v2/pkg/ctx"
+	"github.com/cs3org/reva/v2/pkg/events"
 	"github.com/go-chi/chi/v5"
 	"github.com/go-chi/render"
 	libregraph "github.com/owncloud/libre-graph-api-go"
@@ -133,6 +134,9 @@ func (g Graph) PostUser(w http.ResponseWriter, r *http.Request) {
 		errorcode.GeneralException.Render(w, r, http.StatusInternalServerError, fmt.Sprintf("could not assign role to account %s", err.Error()))
 		return
 	}
+
+	g.publishEvent(events.UserCreated{UserID: *u.Id})
+
 	render.Status(r, http.StatusOK)
 	render.JSON(w, r, u)
 }
@@ -187,6 +191,9 @@ func (g Graph) DeleteUser(w http.ResponseWriter, r *http.Request) {
 			errorcode.GeneralException.Render(w, r, http.StatusInternalServerError, err.Error())
 		}
 	}
+
+	g.publishEvent(events.UserDeleted{UserID: userID})
+
 	render.Status(r, http.StatusNoContent)
 	render.NoContent(w, r)
 }
@@ -211,12 +218,18 @@ func (g Graph) PatchUser(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
+	var features []events.UserFeature
 	if mail, ok := changes.GetMailOk(); ok {
 		if !isValidEmail(*mail) {
 			errorcode.InvalidRequest.Render(w, r, http.StatusBadRequest,
 				fmt.Sprintf("'%s' is not a valid email address", *mail))
 			return
 		}
+		features = append(features, events.UserFeature{Name: "email", Value: *mail})
+	}
+
+	if name, ok := changes.GetDisplayNameOk(); ok {
+		features = append(features, events.UserFeature{Name: "displayname", Value: *name})
 	}
 
 	u, err := g.identityBackend.UpdateUser(r.Context(), nameOrID, *changes)
@@ -229,6 +242,12 @@ func (g Graph) PatchUser(w http.ResponseWriter, r *http.Request) {
 		}
 	}
 
+	g.publishEvent(
+		events.UserFeatureChanged{
+			UserID:   nameOrID,
+			Features: features,
+		},
+	)
 	render.Status(r, http.StatusOK)
 	render.JSON(w, r, u)