check policies before append to queue

application/interfaces.go

@@ -39,6 +39,8 @@ type Actions interface {
 type Invokable interface {
 	// Invoke request, write response. Required header should be set by invoker
 	Invoke(ctx context.Context, request types.Request, response io.Writer, globalEnv map[string]string) error
+	// Unique ID
+	UID() string
 }
 
 // Basic invokable entity
@@ -143,9 +145,15 @@ type Queues interface {
 	Find(targetLambda string) []Queue
 }
 
+type Validator interface {
+	// Inspect request according policy (if applied). Returns null if all checks successful
+	Inspect(lambda string, request *types.Request) error
+}
+
 // Manage policies for the all kind of resource.
 // Lambda can have only one policy at one time, but one policy can be used by many lambdas.
 type Policies interface {
+	Validator
 	// List all policies
 	List() []Policy
 	// Create new policy
@@ -158,6 +166,4 @@ type Policies interface {
 	Apply(lambda string, policy string) error
 	// Clear applied policy for the lambda
 	Clear(lambda string) error
-	// Inspect request according policy (if applied). Returns null if all checks successful
-	Inspect(lambda string, request *types.Request) error
 }

application/lambda/lambda.go

@@ -26,6 +26,8 @@ type localLambda struct {
 	lock      sync.RWMutex
 }
 
+func (local *localLambda) UID() string { return local.uid }
+
 func (local *localLambda) Manifest() types.Manifest {
 	local.lock.RLock()
 	defer local.lock.RUnlock()

application/platform/platform.go

@@ -16,11 +16,7 @@ import (
 
 var allowedName = regexp.MustCompile("^[a-zA-Z0-9._-]{1,255}$")
 
-type Validator interface {
-	Inspect(lambda string, request *types.Request) error
-}
-
-func New(configFile string, validator Validator) (*platform, error) {
+func New(configFile string, validator application.Validator) (*platform, error) {
 	var config application.Config
 	err := config.ReadFile(configFile)
 	if err != nil && !os.IsNotExist(err) {
@@ -40,7 +36,7 @@ type platform struct {
 	config         application.Config
 	configLocation string
 	byUID          map[string]record
-	validator      Validator
+	validator      application.Validator
 }
 
 type record struct {
@@ -205,6 +201,10 @@ func (platform *platform) InvokeByUID(ctx context.Context, uid string, request t
 }
 
 func (platform *platform) Invoke(ctx context.Context, lambda application.Invokable, request types.Request, out io.Writer) error {
+	err := platform.validator.Inspect(lambda.UID(), &request)
+	if err != nil {
+		return err
+	}
 	return lambda.Invoke(ctx, request, out, platform.config.Environment)
 }
 

application/queuemanager/impl.go

@@ -29,10 +29,11 @@ type Platform interface {
 
 type QueueFactory func(name string) (queue.Queue, error)
 
-func New(ctx context.Context, config Store, platform Platform, factory QueueFactory) (*queueManager, error) {
+func New(ctx context.Context, config Store, platform Platform, factory QueueFactory, validator application.Validator) (*queueManager, error) {
 	qm := &queueManager{
 		ctx:          ctx,
 		platform:     platform,
+		validator:    validator,
 		queues:       map[string]*queueDefinition{},
 		queueFactory: factory,
 		config:       config,
@@ -44,6 +45,7 @@ type queueManager struct {
 	ctx          context.Context
 	lock         sync.RWMutex
 	platform     Platform
+	validator    application.Validator
 	queues       map[string]*queueDefinition
 	queueFactory QueueFactory
 	config       Store
@@ -72,6 +74,9 @@ func (qm *queueManager) Put(queue string, request *types.Request) error {
 	if !ok {
 		return fmt.Errorf("queue %s does not exist", queue)
 	}
+	if err := qm.validator.Inspect(q.Target, request); err != nil {
+		return fmt.Errorf("put: security validation failed: %w", err)
+	}
 	return q.queue.Put(qm.ctx, request)
 }
 

application/queuemanager/impl_test.go

@@ -28,6 +28,13 @@ func (mini *mockPlatform) InvokeByUID(ctx context.Context, uid string, request t
 	return handler(request, out)
 }
 
+type bypass struct {
+}
+
+func (b bypass) Inspect(lambda string, request *types.Request) error {
+	return nil
+}
+
 func TestNew(t *testing.T) {
 	var echoText string
 	var echoCh = make(chan struct{})
@@ -58,7 +65,7 @@ func TestNew(t *testing.T) {
 			Target: "greeter",
 		}), platform, func(name string) (queue.Queue, error) {
 			return inmemory.New(10), nil
-		})
+		}, &bypass{})
 	if err != nil {
 		t.Error(err)
 		return

cmd/trusted-cgi/main.go

@@ -146,7 +146,7 @@ func run(ctx context.Context, config Config) error {
 		return err
 	}
 
-	queueManager, err := queuemanager.New(ctx, queuemanager.FileConfig(config.Queues.Config), basePlatform, queueFactory)
+	queueManager, err := queuemanager.New(ctx, queuemanager.FileConfig(config.Queues.Config), basePlatform, queueFactory, policies)
 	if err != nil {
 		return err
 	}

docs/usage/queues.md

@@ -12,7 +12,7 @@ Since `0.3.5`
 
 By default, queues stored in a directory-based style. Each element of queue pipes directly from incoming requests, as well as to lambda without caching. It means - RAM usage is almost constant regardless of request sizes and a number of elements in a queue.
 
-Currently, there are no security restrictions for the queue on append time. All checks will be performed before lambda
+Security restrictions (policies) checked twice: on append time and before lambda
 execution in the same way as it defined in security. 
 
 Queues that bound to the lambda could be found in Overview -> Endpoint page.