Cluster-wide rooms the platform holds

devspace.yaml

@@ -21,6 +21,12 @@ functions:
           containers:
             - name: notifications
               image: ghcr.io/agynio/devcontainer-go:1 # keep in sync with E2E_IMAGE
+              # Both probes go, or nothing ever starts: DevSpace waits for the
+              # pod to be ready before it syncs, the container waits for the
+              # sync before it listens, and the liveness probe kills it at 40s
+              # -- so the sync it is waiting for never arrives.
+              livenessProbe: null
+              readinessProbe: null
               workingDir: /opt/app/data
               command:
                 - sh
@@ -70,8 +76,11 @@ deployments:
           app.kubernetes.io/name: notifications-e2e
 
 commands:
-  test:e2e: |-
-    devspace run-pipeline test:e2e $@
+  # Neither a command nor a pipeline may carry a colon: DevSpace validates both
+  # against ^[a-z0-9][a-z0-9-]*[a-z0-9]$ and refuses to load the whole file when
+  # one does, which took `devspace dev` down with it.
+  test-e2e: |-
+    devspace run-pipeline test-e2e $@
 
 pipelines:
   dev:
@@ -102,7 +111,7 @@ pipelines:
         stop_dev notifications
       fi
 
-  test:e2e:
+  test-e2e:
     run: |-
       create_deployments e2e-runner
       start_dev e2e-runner &

internal/server/authorization.go

@@ -37,12 +37,41 @@ const (
 	identityObjectPrefix     = "identity:"
 	organizationObjectPrefix = "organization:"
 	sandboxObjectPrefix      = "sandbox:"
+	clusterObject            = "cluster:global"
 
 	memberRelation           = "member"
 	canListSandboxesRelation = "can_list_sandboxes"
 	canReadRelation          = "can_read"
+	adminRelation            = "admin"
 )
 
+// platformRoomKinds are the rooms a verified platform caller may hold: the
+// operational state of the resources it places and reaps. Identity-keyed rooms
+// are deliberately absent -- instance_inbox, thread_participant and
+// sandbox_owner belong to a principal, and the platform starts and stops
+// workloads rather than reading what they exchange. A platform caller asking
+// for one of those falls through to the ordinary check, which refuses it.
+var platformRoomKinds = map[roomKind]struct{}{
+	roomKindAgentInstance:  {},
+	roomKindAgent:          {},
+	roomKindSandbox:        {},
+	roomKindSandboxOrg:     {},
+	roomKindWorkload:       {},
+	roomKindVolume:         {},
+	roomKindAgentInstances: {},
+	roomKindSandboxes:      {},
+	roomKindWorkloads:      {},
+}
+
+// requireClusterAdmin settles a platform caller's claim against OpenFGA. The
+// identity type arrives as metadata and is worth nothing on its own; the tuple
+// behind it is written by Identity for exactly one configured id and cannot be
+// requested, so this is the difference between a caller saying it is the
+// platform and it being so.
+func (s *Server) requireClusterAdmin(ctx context.Context, caller uuid.UUID) error {
+	return s.requireRelation(ctx, caller, adminRelation, clusterObject)
+}
+
 // requireOrganizationRelation resolves the caller's relation to an
 // organization. Nothing short of a definite yes admits the subscriber.
 func (s *Server) requireOrganizationRelation(ctx context.Context, caller uuid.UUID, organizationID uuid.UUID, relation string) error {

internal/server/rooms.go

@@ -27,7 +27,17 @@ const (
 	egressRulesRoom             = "egress_rules"
 	llmSubscriptionsRoom        = "llm_subscriptions"
 	environmentsRoom            = "environments"
-	selfRoomIDSegment           = "me"
+	// Flat, and deliberately not keyed by organization. The Orchestrator
+	// reconciles every instance, sandbox and workload in the cluster, so a room set
+	// derived from the organizations it can currently see is a race with the
+	// event that would have told it a new one exists. These three are constant
+	// for the life of the process, so there is nothing to miss and nothing to
+	// re-derive. Unlike the three above they carry every organization's
+	// lifecycle, so they are the platform's alone -- see authorizeSubscribeRooms.
+	agentInstancesRoom = "agent_instances"
+	sandboxesRoom      = "sandboxes"
+	workloadsRoom      = "workloads"
+	selfRoomIDSegment  = "me"
 )
 
 type roomKind int
@@ -48,6 +58,9 @@ const (
 	roomKindEgressRules
 	roomKindLLMSubscriptions
 	roomKindEnvironments
+	roomKindAgentInstances
+	roomKindSandboxes
+	roomKindWorkloads
 )
 
 type subscriptionRoom struct {
@@ -113,6 +126,19 @@ func classifyRoom(room string, callerID uuid.UUID) (roomKind, uuid.UUID, string,
 	if room == environmentsRoom {
 		return roomKindEnvironments, uuid.UUID{}, "", environmentsRoom, nil
 	}
+	// Ahead of the prefix matches below, as the three above are. Neither name
+	// carries the colon its per-entity prefix ends with, so they cannot be
+	// swallowed by one, but settling the literals first keeps that a fact about
+	// this function rather than about the constants.
+	if room == agentInstancesRoom {
+		return roomKindAgentInstances, uuid.UUID{}, "", agentInstancesRoom, nil
+	}
+	if room == sandboxesRoom {
+		return roomKindSandboxes, uuid.UUID{}, "", sandboxesRoom, nil
+	}
+	if room == workloadsRoom {
+		return roomKindWorkloads, uuid.UUID{}, "", workloadsRoom, nil
+	}
 	if strings.HasPrefix(room, threadParticipantRoomPrefix) {
 		raw := strings.TrimSpace(strings.TrimPrefix(room, threadParticipantRoomPrefix))
 		if raw == selfRoomIDSegment {

internal/server/server.go

@@ -126,7 +126,10 @@ type callerIdentity struct {
 	identityType string
 }
 
-const agentInstanceIdentityType = "agent_instance"
+const (
+	agentInstanceIdentityType = "agent_instance"
+	platformIdentityType      = "platform"
+)
 
 func identityIDFromContext(ctx context.Context) (uuid.UUID, error) {
 	md, ok := metadata.FromIncomingContext(ctx)
@@ -174,7 +177,24 @@ func callerIdentityFromContext(ctx context.Context) (callerIdentity, error) {
 // by equality; every other room is gated on the caller's relation to the
 // organization that owns the entity the room reports on.
 func (s *Server) authorizeSubscribeRooms(ctx context.Context, caller callerIdentity, rooms []subscriptionRoom) error {
+	// A platform caller is the platform itself -- the Orchestrator watching what
+	// it places -- rather than a principal acting through it. It used to have no
+	// way to say so, and borrowed an agent instance's id to pass checks written
+	// for that instance. The claim is not taken on faith here: it stands only if
+	// the identity really carries admin on the cluster.
+	platform := false
+	if caller.identityType == platformIdentityType {
+		if err := s.requireClusterAdmin(ctx, caller.id); err != nil {
+			return err
+		}
+		platform = true
+	}
 	for _, room := range rooms {
+		if platform {
+			if _, ok := platformRoomKinds[room.kind]; ok {
+				continue
+			}
+		}
 		switch room.kind {
 		case roomKindThreadParticipant:
 			if room.id != caller.id {
@@ -255,6 +275,12 @@ func (s *Server) authorizeSubscribeRooms(ctx context.Context, caller callerIdent
 			// per-organization room would leave it re-subscribing forever or
 			// missing invalidations for one it has not seen. Bare literals with
 			// no organization to check against.
+		// Reached only by a caller that is not the platform: a verified one
+		// never gets here, having been admitted above. These two carry every
+		// organization's lifecycle in one room, with no id to check a caller
+		// against, so there is no relation that would make them anyone else's.
+		case roomKindAgentInstances, roomKindSandboxes, roomKindWorkloads:
+			return status.Error(codes.PermissionDenied, "permission denied")
 		case roomKindTrace, roomKindVolume:
 			// Both are specified as "member on the owning organization", and
 			// neither owner will say which organization that is: GetTrace

internal/server/server_test.go

@@ -1025,6 +1025,176 @@ func TestSubscribeDeniesAnInstanceAnotherInstancesRoom(t *testing.T) {
 	}
 }
 
+// The Orchestrator watches the sandboxes it reconciles across every
+// organization. It used to borrow an agent instance's identity to do it, which
+// could never satisfy can_list_sandboxes; as itself it holds cluster admin.
+func TestSubscribeSandboxOrgRoomAllowsPlatform(t *testing.T) {
+	t.Parallel()
+
+	hub := stream.NewHub(2, zap.NewNop())
+	client, cleanup := startTestServer(t, &publisherStub{}, hub)
+	defer cleanup()
+
+	ctx, cancel := context.WithTimeout(authenticatedContextWithType(uuid.New(), "platform"), time.Second)
+	defer cancel()
+
+	if _, err := client.Subscribe(ctx, &notificationsv1.SubscribeRequest{
+		Rooms: []string{"sandbox_org:" + uuid.NewString()},
+	}); err != nil {
+		t.Fatalf("Subscribe returned error: %v", err)
+	}
+}
+
+// The identity type is metadata and worth nothing by itself. What admits a
+// platform caller is the cluster admin tuple behind it, so that is what the
+// room must be settled against.
+func TestSubscribePlatformIsCheckedAgainstClusterAdmin(t *testing.T) {
+	t.Parallel()
+
+	callerID := uuid.New()
+	recorder := &recordingAuthz{}
+	client, cleanup := startTestServer(t, &publisherStub{}, &noopHub{},
+		server.WithAuthorization(recorder, allowAll{}, allowAll{}))
+	defer cleanup()
+
+	ctx, cancel := context.WithTimeout(authenticatedContextWithType(callerID, "platform"), 500*time.Millisecond)
+	defer cancel()
+
+	stream, err := client.Subscribe(ctx, &notificationsv1.SubscribeRequest{
+		Rooms: []string{"agent_instance:" + uuid.NewString()},
+	})
+	if err != nil {
+		t.Fatalf("Subscribe returned error: %v", err)
+	}
+	if _, err = stream.Recv(); status.Code(err) == codes.PermissionDenied {
+		t.Fatal("a platform caller holding cluster admin was denied the agent instance room")
+	}
+
+	tuple := recorder.last()
+	if tuple == nil {
+		t.Fatal("expected an authorization check")
+	}
+	if tuple.GetUser() != "identity:"+callerID.String() {
+		t.Fatalf("expected the caller as user, got %q", tuple.GetUser())
+	}
+	if tuple.GetRelation() != "admin" || tuple.GetObject() != "cluster:global" {
+		t.Fatalf("expected admin on cluster:global, got %q on %q", tuple.GetRelation(), tuple.GetObject())
+	}
+}
+
+// Claiming the type is not holding the relation. Anything on the mesh can set
+// the header, so a caller that does not actually carry cluster admin gets
+// nothing from saying it is the platform.
+func TestSubscribePlatformWithoutClusterAdminDenied(t *testing.T) {
+	t.Parallel()
+
+	client, cleanup := startTestServer(t, &publisherStub{}, &noopHub{},
+		server.WithAuthorization(denyAll{}, allowAll{}, allowAll{}))
+	defer cleanup()
+
+	ctx := authenticatedContextWithType(uuid.New(), "platform")
+	streamClient, err := client.Subscribe(ctx, &notificationsv1.SubscribeRequest{
+		Rooms: []string{"sandbox_org:" + uuid.NewString()},
+	})
+	if err == nil {
+		_, err = streamClient.Recv()
+	}
+	if status.Code(err) != codes.PermissionDenied {
+		t.Fatalf("expected PermissionDenied, got %v (%v)", status.Code(err), err)
+	}
+}
+
+// Cluster admin buys the operational rooms, not the private ones. The platform
+// places workloads; it has no business in the inbox one is reading from.
+func TestSubscribePlatformStillDeniedInstanceInbox(t *testing.T) {
+	t.Parallel()
+
+	client, cleanup := startTestServer(t, &publisherStub{}, &noopHub{})
+	defer cleanup()
+
+	ctx := authenticatedContextWithType(uuid.New(), "platform")
+	streamClient, err := client.Subscribe(ctx, &notificationsv1.SubscribeRequest{
+		Rooms: []string{fmt.Sprintf("instance_inbox:%s", uuid.New())},
+	})
+	if err == nil {
+		_, err = streamClient.Recv()
+	}
+	if status.Code(err) != codes.PermissionDenied {
+		t.Fatalf("expected PermissionDenied, got %v (%v)", status.Code(err), err)
+	}
+}
+
+// The rooms the Orchestrator actually holds. Constant names, so its coverage
+// no longer depends on which organizations it managed to enumerate first.
+func TestSubscribeFlatPlatformRoomsAllowPlatform(t *testing.T) {
+	t.Parallel()
+
+	hub := stream.NewHub(2, zap.NewNop())
+	client, cleanup := startTestServer(t, &publisherStub{}, hub)
+	defer cleanup()
+
+	ctx, cancel := context.WithTimeout(authenticatedContextWithType(uuid.New(), "platform"), time.Second)
+	defer cancel()
+
+	if _, err := client.Subscribe(ctx, &notificationsv1.SubscribeRequest{
+		Rooms: []string{"agent_instances", "sandboxes"},
+	}); err != nil {
+		t.Fatalf("Subscribe returned error: %v", err)
+	}
+}
+
+// One room per cluster, carrying every organization's lifecycle. Nobody but the
+// platform has any business holding it, and there is no id in the name to check
+// an ordinary caller against.
+func TestSubscribeFlatPlatformRoomsDenyEveryoneElse(t *testing.T) {
+	t.Parallel()
+
+	for _, room := range []string{"agent_instances", "sandboxes"} {
+		room := room
+		t.Run(room, func(t *testing.T) {
+			t.Parallel()
+
+			client, cleanup := startTestServer(t, &publisherStub{}, &noopHub{})
+			defer cleanup()
+
+			// A permissive authorizer, to show the refusal is the room's own and
+			// not a check that happened to fail.
+			ctx := authenticatedContextWithType(uuid.New(), "agent_instance")
+			streamClient, err := client.Subscribe(ctx, &notificationsv1.SubscribeRequest{Rooms: []string{room}})
+			if err == nil {
+				_, err = streamClient.Recv()
+			}
+			if status.Code(err) != codes.PermissionDenied {
+				t.Fatalf("expected PermissionDenied, got %v (%v)", status.Code(err), err)
+			}
+		})
+	}
+}
+
+// Same boundary, stated for the rooms a person's own client holds.
+func TestSubscribePlatformStillDeniedPrincipalRooms(t *testing.T) {
+	t.Parallel()
+
+	for _, room := range []string{"thread_participant:" + uuid.NewString(), "sandbox_owner:" + uuid.NewString()} {
+		room := room
+		t.Run(room, func(t *testing.T) {
+			t.Parallel()
+
+			client, cleanup := startTestServer(t, &publisherStub{}, &noopHub{})
+			defer cleanup()
+
+			ctx := authenticatedContextWithType(uuid.New(), "platform")
+			streamClient, err := client.Subscribe(ctx, &notificationsv1.SubscribeRequest{Rooms: []string{room}})
+			if err == nil {
+				_, err = streamClient.Recv()
+			}
+			if status.Code(err) != codes.PermissionDenied {
+				t.Fatalf("expected PermissionDenied, got %v (%v)", status.Code(err), err)
+			}
+		})
+	}
+}
+
 func startTestServer(t *testing.T, publisher server.Publisher, hub server.SubscriptionHub, opts ...server.Option) (notificationsv1.NotificationsServiceClient, func()) {
 	t.Helper()