Correct event relationships (#121)

* correct events.nats.token path

Signed-off-by: Mike Mason <mimason@equinix.com>

* add missing loggers

Signed-off-by: Mike Mason <mimason@equinix.com>

* subscriber.Listen is blocking so we never capture and exit with a signal

Signed-off-by: Mike Mason <mimason@equinix.com>

* correct policy example

Signed-off-by: Mike Mason <mimason@equinix.com>

* skip nacking for errors

We need to dig into the errors that may be produced and create a list of
errors which should be reprocessable.

Otherwise we might reprocess a message which will never be succeed,
wasting resources.

Signed-off-by: Mike Mason <mimason@equinix.com>

* create relationships for all matched relations

This allows for relations to be determined based on their type.

All matched relations will be created.

Signed-off-by: Mike Mason <mimason@equinix.com>

---------

Signed-off-by: Mike Mason <mimason@equinix.com>

Author: mikemrm

chart/permissions-api/templates/deployment-worker.yaml

@@ -76,7 +76,7 @@ spec:
           {{- end }}
           {{- if .Values.config.events.nats.token }}
             - name: PERMISSIONSAPI_EVENTS_SUBSCRIBER_NATS_TOKEN
-              value: "{{ .Values.config.events.token }}"
+              value: "{{ .Values.config.events.nats.token }}"
           {{- end }}
           {{- if .Values.config.oidc.issuer }}
           {{- with .Values.config.oidc.audience }}

cmd/worker.go

@@ -62,9 +62,9 @@ func worker(ctx context.Context, cfg *config.AppConfig) {
 		logger.Fatalw("invalid spicedb policy", "error", err)
 	}
 
-	engine := query.NewEngine("infratographer", spiceClient, query.WithPolicy(policy))
+	engine := query.NewEngine("infratographer", spiceClient, query.WithPolicy(policy), query.WithLogger(logger))
 
-	subscriber, err := pubsub.NewSubscriber(ctx, cfg.Events.Subscriber, engine)
+	subscriber, err := pubsub.NewSubscriber(ctx, cfg.Events.Subscriber, engine, pubsub.WithLogger(logger))
 	if err != nil {
 		logger.Fatalw("unable to initialize subscriber", "error", err)
 	}
@@ -75,9 +75,13 @@ func worker(ctx context.Context, cfg *config.AppConfig) {
 		}
 	}
 
-	if err := subscriber.Listen(); err != nil {
-		logger.Fatalw("error listening for events", "error", err)
-	}
+	logger.Info("Listening for events")
+
+	go func() {
+		if err := subscriber.Listen(); err != nil {
+			logger.Fatalw("error listening for events", "error", err)
+		}
+	}()
 
 	// Wait until we're told to stop
 	sig := <-sigCh

internal/pubsub/subscriber.go

@@ -81,6 +81,8 @@ func (s *Subscriber) Subscribe(topic string) error {
 
 	s.changeChannels = append(s.changeChannels, msgChan)
 
+	s.logger.Infof("Subscribing to topic %s", topic)
+
 	return nil
 }
 
@@ -163,6 +165,13 @@ func (s *Subscriber) processEvent(msg *message.Message) error {
 func (s *Subscriber) createRelationships(ctx context.Context, msg *message.Message, resource types.Resource, additionalSubjectIDs []gidx.PrefixedID) error {
 	var relationships []types.Relationship
 
+	rType := s.qe.GetResourceType(resource.Type)
+	if rType == nil {
+		s.logger.Warnw("no resource type found for", "resource_type", resource.Type)
+
+		return nil
+	}
+
 	// Attempt to create relationships from the message fields. If this fails, reject the message
 	for _, id := range additionalSubjectIDs {
 		subjResource, err := s.qe.NewResourceFromID(id)
@@ -172,13 +181,27 @@ func (s *Subscriber) createRelationships(ctx context.Context, msg *message.Messa
 			continue
 		}
 
-		relationship := types.Relationship{
-			Resource: resource,
-			Relation: subjResource.Type,
-			Subject:  subjResource,
-		}
+		for _, rel := range rType.Relationships {
+			var relation string
+
+			for _, tName := range rel.Types {
+				if tName == subjResource.Type {
+					relation = rel.Relation
 
-		relationships = append(relationships, relationship)
+					break
+				}
+			}
+
+			if relation != "" {
+				relationship := types.Relationship{
+					Resource: resource,
+					Relation: relation,
+					Subject:  subjResource,
+				}
+
+				relationships = append(relationships, relationship)
+			}
+		}
 	}
 
 	if len(relationships) == 0 {
@@ -190,9 +213,7 @@ func (s *Subscriber) createRelationships(ctx context.Context, msg *message.Messa
 	// Attempt to create the relationships in SpiceDB. If this fails, nak the message for reprocessing
 	_, err := s.qe.CreateRelationships(ctx, relationships)
 	if err != nil {
-		s.logger.Errorw("error creating relationships - will reprocess", "error", err.Error())
-
-		return err
+		s.logger.Errorw("error creating relationships - will not reprocess", "error", err.Error())
 	}
 
 	return nil
@@ -201,9 +222,7 @@ func (s *Subscriber) createRelationships(ctx context.Context, msg *message.Messa
 func (s *Subscriber) deleteRelationships(ctx context.Context, msg *message.Message, resource types.Resource) error {
 	_, err := s.qe.DeleteRelationships(ctx, resource)
 	if err != nil {
-		s.logger.Errorw("error deleting relationships - will reprocess", "error", err.Error())
-
-		return err
+		s.logger.Errorw("error deleting relationships - will not reprocess", "error", err.Error())
 	}
 
 	return nil

internal/pubsub/subscriber_test.go

@@ -151,7 +151,7 @@ func TestNATS(t *testing.T) {
 				return context.WithValue(ctx, contextKeyEngine, &engine)
 			},
 			CheckFn: func(ctx context.Context, t *testing.T, result testingx.TestResult[*Subscriber]) {
-				require.ErrorIs(t, result.Err, eventtools.ErrNack)
+				require.NoError(t, result.Err)
 			},
 		},
 		{

internal/query/errors.go

@@ -9,4 +9,13 @@ var (
 
 	// ErrInvalidReference represents an error condition where a given SpiceDB object reference is for some reason invalid.
 	ErrInvalidReference = errors.New("invalid reference")
+
+	// ErrInvalidNamespace represents an error when the id prefix is not found in the resource schema
+	ErrInvalidNamespace = errors.New("invalid namespace")
+
+	// ErrInvalidType represents an error when a resource type is not found in the resource schema
+	ErrInvalidType = errors.New("invalid type")
+
+	// ErrInvalidRelationship represents an error when no matching relationship was found
+	ErrInvalidRelationship = errors.New("invalid relationship")
 )

internal/query/mock/mock.go

@@ -104,6 +104,21 @@ func (e *Engine) NewResourceFromID(id gidx.PrefixedID) (types.Resource, error) {
 	return out, nil
 }
 
+// GetResourceType returns the resource type by name
+func (e *Engine) GetResourceType(name string) *types.ResourceType {
+	if e.schema == nil {
+		e.schema = iapl.DefaultPolicy().Schema()
+	}
+
+	for _, resourceType := range e.schema {
+		if resourceType.Name == name {
+			return &resourceType
+		}
+	}
+
+	return nil
+}
+
 // SubjectHasPermission returns nil to satisfy the Engine interface.
 func (e *Engine) SubjectHasPermission(ctx context.Context, subject types.Resource, action string, resource types.Resource, queryToken string) error {
 	e.Called()

internal/query/relations.go

@@ -2,7 +2,6 @@ package query
 
 import (
 	"context"
-	"errors"
 	"io"
 	"strings"
 
@@ -13,20 +12,14 @@ import (
 
 var roleSubjectRelation = "subject"
 
-var (
-	errorInvalidNamespace    = errors.New("invalid namespace")
-	errorInvalidType         = errors.New("invalid type")
-	errorInvalidRelationship = errors.New("invalid relationship")
-)
-
 func (e *engine) getTypeForResource(res types.Resource) (types.ResourceType, error) {
 	for _, resType := range e.schema {
 		if res.Type == resType.Name {
 			return resType, nil
 		}
 	}
 
-	return types.ResourceType{}, errorInvalidType
+	return types.ResourceType{}, ErrInvalidType
 }
 
 func (e *engine) validateRelationship(rel types.Relationship) error {
@@ -40,6 +33,8 @@ func (e *engine) validateRelationship(rel types.Relationship) error {
 		return err
 	}
 
+	e.logger.Infow("validation relationship", "sub", subjType.Name, "rel", rel.Relation, "res", resType.Name)
+
 	for _, typeRel := range resType.Relationships {
 		// If we find a relation with a name and type that matches our relationship,
 		// return
@@ -53,7 +48,7 @@ func (e *engine) validateRelationship(rel types.Relationship) error {
 	}
 
 	// No matching relationship was found, so we should return an error
-	return errorInvalidRelationship
+	return ErrInvalidRelationship
 }
 
 func resourceToSpiceDBRef(namespace string, r types.Resource) *pb.ObjectReference {
@@ -441,7 +436,7 @@ func (e *engine) NewResourceFromID(id gidx.PrefixedID) (types.Resource, error) {
 
 	rType, ok := e.schemaPrefixMap[prefix]
 	if !ok {
-		return types.Resource{}, errorInvalidNamespace
+		return types.Resource{}, ErrInvalidNamespace
 	}
 
 	out := types.Resource{
@@ -451,3 +446,13 @@ func (e *engine) NewResourceFromID(id gidx.PrefixedID) (types.Resource, error) {
 
 	return out, nil
 }
+
+// GetResourceType returns the resource type by name
+func (e *engine) GetResourceType(name string) *types.ResourceType {
+	rType, ok := e.schemaTypeMap[name]
+	if !ok {
+		return nil
+	}
+
+	return &rType
+}

internal/query/relations_test.go

@@ -224,7 +224,7 @@ func TestRelationships(t *testing.T) {
 				Subject:  parentRes,
 			},
 			CheckFn: func(ctx context.Context, t *testing.T, res testingx.TestResult[[]types.Relationship]) {
-				assert.ErrorIs(t, res.Err, errorInvalidRelationship)
+				assert.ErrorIs(t, res.Err, ErrInvalidRelationship)
 			},
 		},
 		{

internal/query/service.go

@@ -21,6 +21,7 @@ type Engine interface {
 	ListRoles(ctx context.Context, resource types.Resource, queryToken string) ([]types.Role, error)
 	DeleteRelationships(ctx context.Context, resource types.Resource) (string, error)
 	NewResourceFromID(id gidx.PrefixedID) (types.Resource, error)
+	GetResourceType(name string) *types.ResourceType
 	SubjectHasPermission(ctx context.Context, subject types.Resource, action string, resource types.Resource, queryToken string) error
 }
 
@@ -30,13 +31,16 @@ type engine struct {
 	client          *authzed.Client
 	schema          []types.ResourceType
 	schemaPrefixMap map[string]types.ResourceType
+	schemaTypeMap   map[string]types.ResourceType
 }
 
-func (e *engine) cacheSchemaPrefixes() {
+func (e *engine) cacheSchemaResources() {
 	e.schemaPrefixMap = make(map[string]types.ResourceType, len(e.schema))
+	e.schemaTypeMap = make(map[string]types.ResourceType, len(e.schema))
 
 	for _, res := range e.schema {
 		e.schemaPrefixMap[res.IDPrefix] = res
+		e.schemaTypeMap[res.Name] = res
 	}
 }
 
@@ -55,7 +59,7 @@ func NewEngine(namespace string, client *authzed.Client, options ...Option) Engi
 	if e.schema == nil {
 		e.schema = iapl.DefaultPolicy().Schema()
 
-		e.cacheSchemaPrefixes()
+		e.cacheSchemaResources()
 	}
 
 	return e
@@ -76,6 +80,6 @@ func WithPolicy(policy iapl.Policy) Option {
 	return func(e *engine) {
 		e.schema = policy.Schema()
 
-		e.cacheSchemaPrefixes()
+		e.cacheSchemaResources()
 	}
 }

policy.example.yaml

@@ -71,13 +71,13 @@ actionbindings:
       - relationshipaction:
           relation: parent
           actionname: loadbalancer_delete
-  - actionname: loadbalancer_create
+  - actionname: loadbalancer_get
     typename: loadbalancer
     conditions:
       - rolebinding: {}
       - relationshipaction:
           relation: owner
-          actionname: loadbalancer_create
+          actionname: loadbalancer_get
   - actionname: loadbalancer_update
     typename: loadbalancer
     conditions: