Added message queue rbac rules

pkg/messagebus/settings.go

@@ -8,6 +8,8 @@ import (
 	"time"
 
 	"github.com/cortezaproject/corteza-server/pkg/filter"
+	"github.com/cortezaproject/corteza-server/pkg/rbac"
+	"github.com/cortezaproject/corteza-server/system/types"
 	"github.com/spf13/cast"
 )
 
@@ -37,11 +39,22 @@ type (
 
 		Deleted filter.State `json:"deleted"`
 
+		// Check fn is called by store backend for each resource found function can
+		// modify the resource and return false if store should not return it
+		//
+		// Store then loads additional resources to satisfy the paging parameters
+		Check func(*QueueSettings) (bool, error) `json:"-"`
+
 		filter.Sorting
 		filter.Paging
 	}
 )
 
+// Resource returns a system resource ID for this type
+func (s QueueSettings) RBACResource() rbac.Resource {
+	return types.MessagebusQueueRBACResource.AppendID(s.ID)
+}
+
 func (h *QueueSettingsMeta) UnmarshalJSON(s []byte) error {
 	type Alias QueueSettingsMeta
 
@@ -53,7 +66,7 @@ func (h *QueueSettingsMeta) UnmarshalJSON(s []byte) error {
 	}
 
 	// set default
-	h.DispatchEvents = true
+	h.DispatchEvents = false
 
 	if err := json.Unmarshal(s, aux); err != nil {
 		return err
@@ -79,7 +92,7 @@ func (m QueueSettingsMeta) MarshalJSON() ([]byte, error) {
 
 	return json.Marshal(struct {
 		PollDelay      string `json:"poll_delay"`
-		DispatchEvents bool   `json:"dispatch_events,omitempty"`
+		DispatchEvents bool   `json:"dispatch_events"`
 	}{
 		PollDelay:      pollDelay,
 		DispatchEvents: m.DispatchEvents,
@@ -102,3 +115,13 @@ func (m *QueueSettingsMeta) Scan(value interface{}) error {
 func (s *QueueSettings) CanDispatch() bool {
 	return s.Meta.DispatchEvents
 }
+
+func ParseQueueSettingsMeta(ss []string) (p QueueSettingsMeta, err error) {
+	p = QueueSettingsMeta{}
+
+	if len(ss) == 0 {
+		return
+	}
+
+	return p, json.Unmarshal([]byte(ss[0]), &p)
+}

provision/000_base/system_access_control.yaml

@@ -14,6 +14,9 @@ allow:
     system:template:
       - render
 
+    system:messagebus-queue:
+      - queue.read
+
   admins:
     system:
       - access
@@ -26,6 +29,7 @@ allow:
       - template.create
       - role.create
       - reminder.assign
+      - messagebus-queue.create
 
     system:application:
       - read
@@ -52,3 +56,10 @@ allow:
       - update
       - delete
       - render
+
+    system:messagebus-queue:
+      - read
+      - update
+      - delete
+      - queue.read
+      - queue.write

store/rdbms/messagebus_queuesettings.go

@@ -9,5 +9,6 @@ import (
 func (s Store) convertMessagebusQueuesettingFilter(f messagebus.QueueSettingsFilter) (query squirrel.SelectBuilder, err error) {
 	query = s.messagebusQueuesettingsSelectBuilder()
 	query = filter.StateCondition(query, "mqs.deleted_at", f.Deleted)
+
 	return
 }

system/rest.yaml

@@ -1479,6 +1479,7 @@ endpoints:
   entrypoint: queues
   path: "/queues"
   imports:
+    - github.com/cortezaproject/corteza-server/pkg/messagebus
     - sqlxTypes github.com/jmoiron/sqlx/types
   apis:
   - name: list
@@ -1537,7 +1538,7 @@ endpoints:
         title: Queue ID
   - name: update
     method: POST
-    title: Update role details
+    title: Update queue details
     path: "/{queueID}"
     parameters:
       path:
@@ -1554,7 +1555,8 @@ endpoints:
         name: handler
         required: true
         title: Queue handler
-      - type: sqlxTypes.JSONText
+      - type: "messagebus.QueueSettingsMeta"
+        parser: "messagebus.ParseQueueSettingsMeta"
         name: meta
         required: false
         title: Meta data for queue

system/rest/queues.go

@@ -8,7 +8,6 @@ import (
 	"github.com/cortezaproject/corteza-server/pkg/messagebus"
 	"github.com/cortezaproject/corteza-server/system/rest/request"
 	"github.com/cortezaproject/corteza-server/system/service"
-	"github.com/davecgh/go-spew/spew"
 )
 
 type (
@@ -51,6 +50,7 @@ func (ctrl *Queue) List(ctx context.Context, r *request.QueuesList) (interface{}
 	}
 
 	set, filter, err := ctrl.svc.Search(ctx, f)
+
 	return ctrl.makeFilterPayload(ctx, set, filter, err)
 }
 
@@ -80,7 +80,7 @@ func (ctrl *Queue) Update(ctx context.Context, r *request.QueuesUpdate) (interfa
 			ID:      r.QueueID,
 			Handler: r.Handler,
 			Queue:   r.Queue,
-			// Meta:    r.Meta,
+			Meta:    r.Meta,
 		}
 	)
 
@@ -90,7 +90,6 @@ func (ctrl *Queue) Update(ctx context.Context, r *request.QueuesUpdate) (interfa
 }
 
 func (ctrl *Queue) Delete(ctx context.Context, r *request.QueuesDelete) (interface{}, error) {
-	spew.Dump("DELETE", r)
 	return api.OK(), ctrl.svc.DeleteByID(ctx, r.QueueID)
 }
 

system/rest/request/queues.go

@@ -11,6 +11,7 @@ package request
 import (
 	"encoding/json"
 	"fmt"
+	"github.com/cortezaproject/corteza-server/pkg/messagebus"
 	"github.com/cortezaproject/corteza-server/pkg/payload"
 	"github.com/go-chi/chi"
 	sqlxTypes "github.com/jmoiron/sqlx/types"
@@ -104,7 +105,7 @@ type (
 		// Meta POST parameter
 		//
 		// Meta data for queue
-		Meta sqlxTypes.JSONText
+		Meta messagebus.QueueSettingsMeta
 	}
 
 	QueuesDelete struct {
@@ -339,7 +340,7 @@ func (r QueuesUpdate) GetHandler() string {
 }
 
 // Auditable returns all auditable/loggable parameters
-func (r QueuesUpdate) GetMeta() sqlxTypes.JSONText {
+func (r QueuesUpdate) GetMeta() messagebus.QueueSettingsMeta {
 	return r.Meta
 }
 
@@ -378,8 +379,13 @@ func (r *QueuesUpdate) Fill(req *http.Request) (err error) {
 			}
 		}
 
-		if val, ok := req.Form["meta"]; ok && len(val) > 0 {
+		if val, ok := req.Form["meta[]"]; ok {
-			r.Meta, err = payload.ParseJSONTextWithErr(val[0])
+			r.Meta, err = messagebus.ParseQueueSettingsMeta(val)
+			if err != nil {
+				return err
+			}
+		} else if val, ok := req.Form["meta"]; ok {
+			r.Meta, err = messagebus.ParseQueueSettingsMeta(val)
 			if err != nil {
 				return err
 			}

system/service/access_control.go

@@ -5,6 +5,7 @@ import (
 
 	"github.com/cortezaproject/corteza-server/pkg/actionlog"
 	internalAuth "github.com/cortezaproject/corteza-server/pkg/auth"
+	"github.com/cortezaproject/corteza-server/pkg/messagebus"
 
 	"github.com/cortezaproject/corteza-server/pkg/rbac"
 	"github.com/cortezaproject/corteza-server/system/types"
@@ -47,6 +48,7 @@ func (svc accessControl) Effective(ctx context.Context) (ee rbac.EffectiveSet) {
 	ee.Push(types.SystemRBACResource, "application.flag.global", svc.CanGlobalFlagApplication(ctx))
 	ee.Push(types.SystemRBACResource, "template.create", svc.CanCreateTemplate(ctx))
 	ee.Push(types.SystemRBACResource, "role.create", svc.CanCreateRole(ctx))
+	ee.Push(types.SystemRBACResource, "messagebus-queue.create", svc.CanCreateMessagebusQueue(ctx))
 
 	return
 }
@@ -95,6 +97,10 @@ func (svc accessControl) CanAssignReminder(ctx context.Context) bool {
 	return svc.can(ctx, types.SystemRBACResource, "reminder.assign")
 }
 
+func (svc accessControl) CanCreateMessagebusQueue(ctx context.Context) bool {
+	return svc.can(ctx, types.SystemRBACResource, "messagebus-queue.create")
+}
+
 func (svc accessControl) CanReadRole(ctx context.Context, rl *types.Role) bool {
 	return svc.can(ctx, rl.RBACResource(), "read", rbac.Allowed)
 }
@@ -208,6 +214,26 @@ func (svc accessControl) CanUnmaskName(ctx context.Context, u *types.User) bool
 	return svc.can(ctx, u.RBACResource(), "unmask.name")
 }
 
+func (svc accessControl) CanReadMessagebusQueue(ctx context.Context, c *messagebus.QueueSettings) bool {
+	return svc.can(ctx, c.RBACResource(), "read")
+}
+
+func (svc accessControl) CanUpdateMessagebusQueue(ctx context.Context, c *messagebus.QueueSettings) bool {
+	return svc.can(ctx, c.RBACResource(), "update")
+}
+
+func (svc accessControl) CanDeleteMessagebusQueue(ctx context.Context, c *messagebus.QueueSettings) bool {
+	return svc.can(ctx, c.RBACResource(), "delete")
+}
+
+func (svc accessControl) CanReadFromMessagebusQueue(ctx context.Context, c *messagebus.QueueSettings) bool {
+	return svc.can(ctx, c.RBACResource(), "queue.read")
+}
+
+func (svc accessControl) CanWriteToMessagebusQueue(ctx context.Context, c *messagebus.QueueSettings) bool {
+	return svc.can(ctx, c.RBACResource(), "queue.write")
+}
+
 func (svc accessControl) can(ctx context.Context, res rbac.Resource, op rbac.Operation, ff ...rbac.CheckAccessFunc) bool {
 	var (
 		u     = internalAuth.GetIdentityFromContext(ctx)
@@ -276,6 +302,7 @@ func (svc accessControl) Whitelist() rbac.Whitelist {
 		"application.flag.global",
 		"template.create",
 		"reminder.assign",
+		"messagebus-queue.create",
 	)
 
 	wl.Set(
@@ -321,5 +348,14 @@ func (svc accessControl) Whitelist() rbac.Whitelist {
 		"authorize",
 	)
 
+	wl.Set(
+		types.MessagebusQueueRBACResource,
+		"read",
+		"update",
+		"delete",
+		"queue.read",
+		"queue.write",
+	)
+
 	return wl
 }

system/service/queue.go

@@ -12,7 +12,7 @@ type (
 	queue struct {
 		actionlog actionlog.Recorder
 		store     store.Storer
-		ac        templateAccessController
+		ac        queueAccessController
 	}
 
 	QueueService interface {
@@ -25,6 +25,16 @@ type (
 		DeleteByID(ctx context.Context, ID uint64) error
 		UndeleteByID(ctx context.Context, ID uint64) error
 	}
+
+	queueAccessController interface {
+		CanCreateMessagebusQueue(ctx context.Context) bool
+		CanReadMessagebusQueue(ctx context.Context, c *messagebus.QueueSettings) bool
+		CanUpdateMessagebusQueue(ctx context.Context, c *messagebus.QueueSettings) bool
+		CanDeleteMessagebusQueue(ctx context.Context, c *messagebus.QueueSettings) bool
+
+		CanReadFromMessagebusQueue(ctx context.Context, c *messagebus.QueueSettings) bool
+		CanWriteToMessagebusQueue(ctx context.Context, c *messagebus.QueueSettings) bool
+	}
 )
 
 func Queue() QueueService {
@@ -51,9 +61,9 @@ func (svc *queue) FindByID(ctx context.Context, ID uint64) (q *messagebus.QueueS
 
 		qProps.setQueue(q)
 
-		// if !svc.ac.CanReadTemplate(ctx, tpl) {
+		if !svc.ac.CanReadMessagebusQueue(ctx, q) {
-		// 	return TemplateErrNotAllowedToRead()
+			return QueueErrNotAllowedToRead(qProps)
-		// }
+		}
 
 		return nil
 	}()
@@ -67,6 +77,10 @@ func (svc *queue) Create(ctx context.Context, new *messagebus.QueueSettings) (q
 	)
 
 	err = func() (err error) {
+		if !svc.ac.CanCreateMessagebusQueue(ctx) {
+			return QueueErrNotAllowedToCreate(qProps)
+		}
+
 		if !svc.isValidHandler(messagebus.HandlerType(new.Handler)) {
 			return QueueErrInvalidHandler(qProps)
 		}
@@ -90,10 +104,16 @@ func (svc *queue) Create(ctx context.Context, new *messagebus.QueueSettings) (q
 func (svc *queue) Update(ctx context.Context, upd *messagebus.QueueSettings) (q *messagebus.QueueSettings, err error) {
 	var (
 		qProps = &queueActionProps{update: upd}
+		qq     *messagebus.QueueSettings
+		e      error
 	)
 
 	err = func() (err error) {
-		if _, e := store.LookupMessagebusQueuesettingByID(ctx, svc.store, upd.ID); e != nil {
+		if !svc.ac.CanUpdateMessagebusQueue(ctx, upd) {
+			return QueueErrNotAllowedToUpdate(qProps)
+		}
+
+		if qq, e = store.LookupMessagebusQueuesettingByID(ctx, svc.store, upd.ID); e != nil {
 			return QueueErrNotFound(qProps)
 		}
 
@@ -107,6 +127,7 @@ func (svc *queue) Update(ctx context.Context, upd *messagebus.QueueSettings) (q
 
 		// Set new values after beforeCreate events are emitted
 		upd.UpdatedAt = now()
+		upd.CreatedAt = qq.CreatedAt
 
 		if err = store.UpdateMessagebusQueuesetting(ctx, svc.store, upd); err != nil {
 			return
@@ -137,9 +158,9 @@ func (svc *queue) DeleteByID(ctx context.Context, ID uint64) (err error) {
 
 		qProps.setQueue(q)
 
-		// if !svc.ac.CanDeleteTemplate(ctx, tpl) {
+		if !svc.ac.CanDeleteMessagebusQueue(ctx, q) {
-		// 	return TemplateErrNotAllowedToDelete()
+			return QueueErrNotAllowedToDelete(qProps)
-		// }
+		}
 
 		q.DeletedAt = now()
 		if err = store.UpdateMessagebusQueuesetting(ctx, svc.store, q); err != nil {
@@ -169,9 +190,9 @@ func (svc *queue) UndeleteByID(ctx context.Context, ID uint64) (err error) {
 
 		qProps.setQueue(q)
 
-		// if !svc.ac.CanDeleteTemplate(ctx, tpl) {
+		if !svc.ac.CanDeleteMessagebusQueue(ctx, q) {
-		// 	return TemplateErrNotAllowedToDelete()
+			return QueueErrNotAllowedToDelete(qProps)
-		// }
+		}
 
 		q.DeletedAt = nil
 		if err = store.UpdateMessagebusQueuesetting(ctx, svc.store, q); err != nil {
@@ -189,6 +210,15 @@ func (svc *queue) Search(ctx context.Context, filter messagebus.QueueSettingsFil
 		aProps = &queueActionProps{search: &filter}
 	)
 
+	// For each fetched item, store backend will check if it is valid or not
+	filter.Check = func(res *messagebus.QueueSettings) (bool, error) {
+		if !svc.ac.CanReadMessagebusQueue(ctx, res) {
+			return false, nil
+		}
+
+		return true, nil
+	}
+
 	err = func() error {
 		if q, f, err = store.SearchMessagebusQueuesettings(ctx, svc.store, filter); err != nil {
 			return err

system/service/queue_actions.gen.go

@@ -514,6 +514,38 @@ func QueueErrAlreadyExists(mm ...*queueActionProps) *errors.Error {
 	return e
 }
 
+// QueueErrNotAllowedToCreate returns "system:queue.notAllowedToCreate" as *errors.Error
+//
+//
+// This function is auto-generated.
+//
+func QueueErrNotAllowedToCreate(mm ...*queueActionProps) *errors.Error {
+	var p = &queueActionProps{}
+	if len(mm) > 0 {
+		p = mm[0]
+	}
+
+	var e = errors.New(
+		errors.KindInternal,
+
+		p.Format("not allowed to create a queue", nil),
+
+		errors.Meta("type", "notAllowedToCreate"),
+		errors.Meta("resource", "system:queue"),
+
+		// action log entry; no formatting, it will be applied inside recordAction fn.
+		errors.Meta(queueLogMetaKey{}, "failed to create a queue; insufficient permissions"),
+		errors.Meta(queuePropsMetaKey{}, p),
+
+		errors.StackSkip(1),
+	)
+
+	if len(mm) > 0 {
+	}
+
+	return e
+}
+
 // QueueErrNotAllowedToRead returns "system:queue.notAllowedToRead" as *errors.Error
 //
 //
@@ -546,6 +578,166 @@ func QueueErrNotAllowedToRead(mm ...*queueActionProps) *errors.Error {
 	return e
 }
 
+// QueueErrNotAllowedToUpdate returns "system:queue.notAllowedToUpdate" as *errors.Error
+//
+//
+// This function is auto-generated.
+//
+func QueueErrNotAllowedToUpdate(mm ...*queueActionProps) *errors.Error {
+	var p = &queueActionProps{}
+	if len(mm) > 0 {
+		p = mm[0]
+	}
+
+	var e = errors.New(
+		errors.KindInternal,
+
+		p.Format("not allowed to update this queue", nil),
+
+		errors.Meta("type", "notAllowedToUpdate"),
+		errors.Meta("resource", "system:queue"),
+
+		// action log entry; no formatting, it will be applied inside recordAction fn.
+		errors.Meta(queueLogMetaKey{}, "failed to update {queue.queue}; insufficient permissions"),
+		errors.Meta(queuePropsMetaKey{}, p),
+
+		errors.StackSkip(1),
+	)
+
+	if len(mm) > 0 {
+	}
+
+	return e
+}
+
+// QueueErrNotAllowedToDelete returns "system:queue.notAllowedToDelete" as *errors.Error
+//
+//
+// This function is auto-generated.
+//
+func QueueErrNotAllowedToDelete(mm ...*queueActionProps) *errors.Error {
+	var p = &queueActionProps{}
+	if len(mm) > 0 {
+		p = mm[0]
+	}
+
+	var e = errors.New(
+		errors.KindInternal,
+
+		p.Format("not allowed to delete this queue", nil),
+
+		errors.Meta("type", "notAllowedToDelete"),
+		errors.Meta("resource", "system:queue"),
+
+		// action log entry; no formatting, it will be applied inside recordAction fn.
+		errors.Meta(queueLogMetaKey{}, "failed to delete {queue.queue}; insufficient permissions"),
+		errors.Meta(queuePropsMetaKey{}, p),
+
+		errors.StackSkip(1),
+	)
+
+	if len(mm) > 0 {
+	}
+
+	return e
+}
+
+// QueueErrNotAllowedToUndelete returns "system:queue.notAllowedToUndelete" as *errors.Error
+//
+//
+// This function is auto-generated.
+//
+func QueueErrNotAllowedToUndelete(mm ...*queueActionProps) *errors.Error {
+	var p = &queueActionProps{}
+	if len(mm) > 0 {
+		p = mm[0]
+	}
+
+	var e = errors.New(
+		errors.KindInternal,
+
+		p.Format("not allowed to undelete this queue", nil),
+
+		errors.Meta("type", "notAllowedToUndelete"),
+		errors.Meta("resource", "system:queue"),
+
+		// action log entry; no formatting, it will be applied inside recordAction fn.
+		errors.Meta(queueLogMetaKey{}, "failed to undelete {queue.queue}; insufficient permissions"),
+		errors.Meta(queuePropsMetaKey{}, p),
+
+		errors.StackSkip(1),
+	)
+
+	if len(mm) > 0 {
+	}
+
+	return e
+}
+
+// QueueErrNotAllowedToWriteTo returns "system:queue.notAllowedToWriteTo" as *errors.Error
+//
+//
+// This function is auto-generated.
+//
+func QueueErrNotAllowedToWriteTo(mm ...*queueActionProps) *errors.Error {
+	var p = &queueActionProps{}
+	if len(mm) > 0 {
+		p = mm[0]
+	}
+
+	var e = errors.New(
+		errors.KindInternal,
+
+		p.Format("not allowed to add messages to this queue", nil),
+
+		errors.Meta("type", "notAllowedToWriteTo"),
+		errors.Meta("resource", "system:queue"),
+
+		// action log entry; no formatting, it will be applied inside recordAction fn.
+		errors.Meta(queueLogMetaKey{}, "failed to add message to {queue.queue}; insufficient permissions"),
+		errors.Meta(queuePropsMetaKey{}, p),
+
+		errors.StackSkip(1),
+	)
+
+	if len(mm) > 0 {
+	}
+
+	return e
+}
+
+// QueueErrNotAllowedToReadFrom returns "system:queue.notAllowedToReadFrom" as *errors.Error
+//
+//
+// This function is auto-generated.
+//
+func QueueErrNotAllowedToReadFrom(mm ...*queueActionProps) *errors.Error {
+	var p = &queueActionProps{}
+	if len(mm) > 0 {
+		p = mm[0]
+	}
+
+	var e = errors.New(
+		errors.KindInternal,
+
+		p.Format("not allowed to read messages from this queue", nil),
+
+		errors.Meta("type", "notAllowedToReadFrom"),
+		errors.Meta("resource", "system:queue"),
+
+		// action log entry; no formatting, it will be applied inside recordAction fn.
+		errors.Meta(queueLogMetaKey{}, "failed to read message from {queue.queue}; insufficient permissions"),
+		errors.Meta(queuePropsMetaKey{}, p),
+
+		errors.StackSkip(1),
+	)
+
+	if len(mm) > 0 {
+	}
+
+	return e
+}
+
 // *********************************************************************************************************************
 // *********************************************************************************************************************
 

system/service/queue_actions.yaml

@@ -64,26 +64,30 @@ errors:
     message: "queue by that name already exists"
     severity: warning
 
+  - error: notAllowedToCreate
+    message: "not allowed to create a queue"
+    log: "failed to create a queue; insufficient permissions"
+
   - error: notAllowedToRead
     message: "not allowed to read this queue"
     log: "failed to read {queue.queue}; insufficient permissions"
 
-  # - error: notAllowedToCreate
+  - error: notAllowedToUpdate
-  #   message: "not allowed to create templates"
+    message: "not allowed to update this queue"
-  #   log: "failed to create template; insufficient permissions"
+    log: "failed to update {queue.queue}; insufficient permissions"
 
-  # - error: notAllowedToUpdate
+  - error: notAllowedToDelete
-  #   message: "not allowed to update this template"
+    message: "not allowed to delete this queue"
-  #   log: "failed to update {template.handle}; insufficient permissions"
+    log: "failed to delete {queue.queue}; insufficient permissions"
 
-  # - error: notAllowedToDelete
+  - error: notAllowedToUndelete
-  #   message: "not allowed to delete this template"
+    message: "not allowed to undelete this queue"
-  #   log: "failed to delete {template.handle}; insufficient permissions"
+    log: "failed to undelete {queue.queue}; insufficient permissions"
 
-  # - error: notAllowedToUndelete
+  - error: notAllowedToWriteTo
-  #   message: "not allowed to undelete this template"
+    message: "not allowed to add messages to this queue"
-  #   log: "failed to undelete {template.handle}; insufficient permissions"
+    log: "failed to add message to {queue.queue}; insufficient permissions"
 
-  # - error: notAllowedToRender
+  - error: notAllowedToReadFrom
-  #   message: "not allowed to render this template"
+    message: "not allowed to read messages from this queue"
-  #   log: "failed to render {template.handle}; insufficient permissions"
+    log: "failed to read message from {queue.queue}; insufficient permissions"

system/types/permission_resources.go

@@ -10,3 +10,4 @@ const TemplateRBACResource = rbac.Resource("system:template:")
 const UserRBACResource = rbac.Resource("system:user:")
 const RoleRBACResource = rbac.Resource("system:role:")
 const AuthClientRBACResource = rbac.Resource("system:auth-client:")
+const MessagebusQueueRBACResource = rbac.Resource("system:messagebus-queue:")