[scopes] feat: add sub_kind for scoped role assignments (#64752)

* [scopes] feat: add sub_kind for scoped role assignments

* make unknown subkinds readable but not writable

api/gen/proto/go/teleport/scopes/access/v1/service.pb.go

@@ -536,7 +536,9 @@ func (*DeleteScopedRoleResponse) Descriptor() ([]byte, []int) {
 type GetScopedRoleAssignmentRequest struct {
 	state protoimpl.MessageState `protogen:"open.v1"`
 	// Name is the name of the scoped role assignment.
-	Name          string `protobuf:"bytes,1,opt,name=name,proto3" json:"name,omitempty"`
+	Name string `protobuf:"bytes,1,opt,name=name,proto3" json:"name,omitempty"`
+	// SubKind is the sub kind of the scoped role assignment.
+	SubKind       string `protobuf:"bytes,2,opt,name=sub_kind,json=subKind,proto3" json:"sub_kind,omitempty"`
 	unknownFields protoimpl.UnknownFields
 	sizeCache     protoimpl.SizeCache
 }
@@ -578,6 +580,13 @@ func (x *GetScopedRoleAssignmentRequest) GetName() string {
 	return ""
 }
 
+func (x *GetScopedRoleAssignmentRequest) GetSubKind() string {
+	if x != nil {
+		return x.SubKind
+	}
+	return ""
+}
+
 // GetScopedRoleAssignmentResponse is the response to get a scoped role assignment.
 type GetScopedRoleAssignmentResponse struct {
 	state protoimpl.MessageState `protogen:"open.v1"`
@@ -890,7 +899,9 @@ type DeleteScopedRoleAssignmentRequest struct {
 	// Name is the name of the scoped role assignment to delete.
 	Name string `protobuf:"bytes,1,opt,name=name,proto3" json:"name,omitempty"`
 	// Revision asserts the revision of the scoped role assignment to delete (optional).
-	Revision      string `protobuf:"bytes,2,opt,name=revision,proto3" json:"revision,omitempty"`
+	Revision string `protobuf:"bytes,2,opt,name=revision,proto3" json:"revision,omitempty"`
+	// SubKind is the sub kind of the scoped role assignment.
+	SubKind       string `protobuf:"bytes,3,opt,name=sub_kind,json=subKind,proto3" json:"sub_kind,omitempty"`
 	unknownFields protoimpl.UnknownFields
 	sizeCache     protoimpl.SizeCache
 }
@@ -939,6 +950,13 @@ func (x *DeleteScopedRoleAssignmentRequest) GetRevision() string {
 	return ""
 }
 
+func (x *DeleteScopedRoleAssignmentRequest) GetSubKind() string {
+	if x != nil {
+		return x.SubKind
+	}
+	return ""
+}
+
 // DeleteScopedRoleAssignmentResponse is the response to delete a scoped role assignment.
 type DeleteScopedRoleAssignmentResponse struct {
 	state         protoimpl.MessageState `protogen:"open.v1"`
@@ -1005,9 +1023,10 @@ const file_teleport_scopes_access_v1_service_proto_rawDesc = "" +
 	"\x17DeleteScopedRoleRequest\x12\x12\n" +
 	"\x04name\x18\x01 \x01(\tR\x04name\x12\x1a\n" +
 	"\brevision\x18\x02 \x01(\tR\brevision\"\x1a\n" +
-	"\x18DeleteScopedRoleResponse\"4\n" +
+	"\x18DeleteScopedRoleResponse\"O\n" +
 	"\x1eGetScopedRoleAssignmentRequest\x12\x12\n" +
-	"\x04name\x18\x01 \x01(\tR\x04name\"r\n" +
+	"\x04name\x18\x01 \x01(\tR\x04name\x12\x19\n" +
+	"\bsub_kind\x18\x02 \x01(\tR\asubKind\"r\n" +
 	"\x1fGetScopedRoleAssignmentResponse\x12O\n" +
 	"\n" +
 	"assignment\x18\x01 \x01(\v2/.teleport.scopes.access.v1.ScopedRoleAssignmentR\n" +
@@ -1035,10 +1054,11 @@ const file_teleport_scopes_access_v1_service_proto_rawDesc = "" +
 	"\"CreateScopedRoleAssignmentResponse\x12O\n" +
 	"\n" +
 	"assignment\x18\x01 \x01(\v2/.teleport.scopes.access.v1.ScopedRoleAssignmentR\n" +
-	"assignment\"S\n" +
+	"assignment\"n\n" +
 	"!DeleteScopedRoleAssignmentRequest\x12\x12\n" +
 	"\x04name\x18\x01 \x01(\tR\x04name\x12\x1a\n" +
-	"\brevision\x18\x02 \x01(\tR\brevision\"$\n" +
+	"\brevision\x18\x02 \x01(\tR\brevision\x12\x19\n" +
+	"\bsub_kind\x18\x03 \x01(\tR\asubKind\"$\n" +
 	"\"DeleteScopedRoleAssignmentResponse2\xde\t\n" +
 	"\x13ScopedAccessService\x12r\n" +
 	"\rGetScopedRole\x12/.teleport.scopes.access.v1.GetScopedRoleRequest\x1a0.teleport.scopes.access.v1.GetScopedRoleResponse\x12x\n" +

api/proto/teleport/scopes/access/v1/service.proto

@@ -130,6 +130,8 @@ message DeleteScopedRoleResponse {}
 message GetScopedRoleAssignmentRequest {
   // Name is the name of the scoped role assignment.
   string name = 1;
+  // SubKind is the sub kind of the scoped role assignment.
+  string sub_kind = 2;
 }
 
 // GetScopedRoleAssignmentResponse is the response to get a scoped role assignment.
@@ -197,6 +199,9 @@ message DeleteScopedRoleAssignmentRequest {
 
   // Revision asserts the revision of the scoped role assignment to delete (optional).
   string revision = 2;
+
+  // SubKind is the sub kind of the scoped role assignment.
+  string sub_kind = 3;
 }
 
 // DeleteScopedRoleAssignmentResponse is the response to delete a scoped role assignment.

lib/auth/auth_login_test.go

@@ -1047,7 +1047,8 @@ func TestBasicSSHScopedLogin(t *testing.T) {
 		assignmentIDs = append(assignmentIDs, assignmentID)
 		_, err = adminClient.ScopedAccessServiceClient().CreateScopedRoleAssignment(ctx, &scopedaccessv1.CreateScopedRoleAssignmentRequest{
 			Assignment: &scopedaccessv1.ScopedRoleAssignment{
-				Kind: scopedaccess.KindScopedRoleAssignment,
+				Kind:    scopedaccess.KindScopedRoleAssignment,
+				SubKind: scopedaccess.SubKindDynamic,
 				Metadata: &headerv1.Metadata{
 					Name: assignmentID,
 				},

lib/auth/auth_with_roles_test.go

@@ -10295,7 +10295,8 @@ func TestScopedRoleEvents(t *testing.T) {
 	_ = getNextEvent() // drain the role create event
 
 	assignment := &scopedaccessv1.ScopedRoleAssignment{
-		Kind: scopedaccess.KindScopedRoleAssignment,
+		Kind:    scopedaccess.KindScopedRoleAssignment,
+		SubKind: scopedaccess.SubKindDynamic,
 		Metadata: &headerv1.Metadata{
 			Name: uuid.New().String(),
 		},
@@ -10327,7 +10328,8 @@ func TestScopedRoleEvents(t *testing.T) {
 
 	// delete the assignment and verify delete event is well-formed.
 	_, err = service.DeleteScopedRoleAssignment(ctx, &scopedaccessv1.DeleteScopedRoleAssignmentRequest{
-		Name: assignment.Metadata.Name,
+		Name:    assignment.Metadata.Name,
+		SubKind: assignment.SubKind,
 	})
 	require.NoError(t, err)
 
@@ -10335,7 +10337,8 @@ func TestScopedRoleEvents(t *testing.T) {
 	require.Equal(t, types.OpDelete, event.Type)
 
 	require.Empty(t, cmp.Diff(&types.ResourceHeader{
-		Kind: scopedaccess.KindScopedRoleAssignment,
+		Kind:    scopedaccess.KindScopedRoleAssignment,
+		SubKind: scopedaccess.SubKindDynamic,
 		Metadata: types.Metadata{
 			Name: assignment.Metadata.Name,
 		},

lib/auth/scopes/access/service.go

@@ -278,7 +278,8 @@ func (s *Server) DeleteScopedRoleAssignment(ctx context.Context, req *scopedacce
 
 	// load the assignment so we can determine the resource scope
 	grsp, err := s.cfg.Reader.GetScopedRoleAssignment(ctx, &scopedaccessv1.GetScopedRoleAssignmentRequest{
-		Name: req.GetName(),
+		Name:    req.GetName(),
+		SubKind: req.GetSubKind(),
 	})
 	if err != nil {
 		if trace.IsNotFound(err) {

lib/auth/scopes/access/service_test.go

@@ -166,9 +166,7 @@ func newBackendPack(t *testing.T) *backendPack {
 func TestRoleBasics(t *testing.T) {
 	t.Setenv("TELEPORT_UNSTABLE_SCOPES", "yes")
 
-	ctx, cancel := context.WithCancel(context.Background())
-	defer cancel()
-
+	ctx := t.Context()
 	bk := newBackendPack(t)
 	defer bk.Close()
 
@@ -418,8 +416,7 @@ func TestRoleBasics(t *testing.T) {
 func TestAssignmentBasics(t *testing.T) {
 	t.Setenv("TELEPORT_UNSTABLE_SCOPES", "yes")
 
-	ctx, cancel := context.WithCancel(context.Background())
-	defer cancel()
+	ctx := t.Context()
 
 	bk := newBackendPack(t)
 	defer bk.Close()
@@ -508,7 +505,8 @@ func TestAssignmentBasics(t *testing.T) {
 
 	// verify expected successful read
 	rasp, err := srv.GetScopedRoleAssignment(ctx, &scopedaccessv1.GetScopedRoleAssignmentRequest{
-		Name: initialAssignments[0].GetMetadata().GetName(),
+		Name:    initialAssignments[0].GetMetadata().GetName(),
+		SubKind: initialAssignments[0].GetSubKind(),
 	})
 	require.NoError(t, err)
 	require.Equal(t, initialAssignments[0].GetMetadata().GetName(), rasp.GetAssignment().GetMetadata().GetName())
@@ -516,7 +514,8 @@ func TestAssignmentBasics(t *testing.T) {
 
 	// verify expected denied read
 	rasp, err = srv.GetScopedRoleAssignment(ctx, &scopedaccessv1.GetScopedRoleAssignmentRequest{
-		Name: initialAssignments[1].GetMetadata().GetName(),
+		Name:    initialAssignments[1].GetMetadata().GetName(),
+		SubKind: initialAssignments[1].GetSubKind(),
 	})
 	require.Error(t, err)
 	require.True(t, trace.IsAccessDenied(err), "expected access denied error, got: %v", err)
@@ -554,7 +553,8 @@ func TestAssignmentBasics(t *testing.T) {
 	// verify that denied create really didn't create the assignment (requires using backend service
 	// directly to avoid false positive due to cache replication)
 	garsp, err := bk.service.GetScopedRoleAssignment(ctx, &scopedaccessv1.GetScopedRoleAssignmentRequest{
-		Name: a2.GetMetadata().GetName(),
+		Name:    a2.GetMetadata().GetName(),
+		SubKind: a2.GetSubKind(),
 	})
 	require.Error(t, err)
 	require.True(t, trace.IsNotFound(err), "expected not found error, got: %v", err)
@@ -562,7 +562,8 @@ func TestAssignmentBasics(t *testing.T) {
 
 	// verify expected successful delete
 	_, err = srv.DeleteScopedRoleAssignment(ctx, &scopedaccessv1.DeleteScopedRoleAssignmentRequest{
-		Name: a1.GetMetadata().GetName(),
+		Name:    a1.GetMetadata().GetName(),
+		SubKind: a1.GetSubKind(),
 	})
 	require.NoError(t, err)
 
@@ -578,7 +579,8 @@ func TestAssignmentBasics(t *testing.T) {
 
 	// verify expected denied delete (out of scope)
 	_, err = srv.DeleteScopedRoleAssignment(ctx, &scopedaccessv1.DeleteScopedRoleAssignmentRequest{
-		Name: initialAssignments[1].GetMetadata().GetName(),
+		Name:    initialAssignments[1].GetMetadata().GetName(),
+		SubKind: initialAssignments[1].GetSubKind(),
 	})
 	require.Error(t, err)
 	require.True(t, trace.IsAccessDenied(err), "expected access denied error, got: %v", err)
@@ -586,7 +588,8 @@ func TestAssignmentBasics(t *testing.T) {
 	// verify that denied delete really didn't delete the assignment (requires using backend service
 	// directly to avoid false positive due to cache replication)
 	rasp, err = bk.service.GetScopedRoleAssignment(ctx, &scopedaccessv1.GetScopedRoleAssignmentRequest{
-		Name: initialAssignments[1].GetMetadata().GetName(),
+		Name:    initialAssignments[1].GetMetadata().GetName(),
+		SubKind: initialAssignments[1].GetSubKind(),
 	})
 	require.NoError(t, err)
 	require.Equal(t, initialAssignments[1].GetMetadata().GetName(), rasp.GetAssignment().GetMetadata().GetName())
@@ -594,7 +597,8 @@ func TestAssignmentBasics(t *testing.T) {
 
 func newScopedRoleAssignmentAtScope(roleName string, scope string) *scopedaccessv1.ScopedRoleAssignment {
 	return &scopedaccessv1.ScopedRoleAssignment{
-		Kind: scopedaccess.KindScopedRoleAssignment,
+		Kind:    scopedaccess.KindScopedRoleAssignment,
+		SubKind: scopedaccess.SubKindDynamic,
 		Metadata: &headerv1.Metadata{
 			Name: uuid.New().String(),
 		},
@@ -616,9 +620,7 @@ func newScopedRoleAssignmentAtScope(roleName string, scope string) *scopedaccess
 func TestUnscopedBasics(t *testing.T) {
 	t.Setenv("TELEPORT_UNSTABLE_SCOPES", "yes")
 
-	ctx, cancel := context.WithCancel(context.Background())
-	defer cancel()
-
+	ctx := t.Context()
 	bk := newBackendPack(t)
 	defer bk.Close()
 
@@ -743,7 +745,8 @@ func TestUnscopedBasics(t *testing.T) {
 	// verify that admin can create an assignment
 	acrsp, err := srvAlice.CreateScopedRoleAssignment(ctx, &scopedaccessv1.CreateScopedRoleAssignmentRequest{
 		Assignment: &scopedaccessv1.ScopedRoleAssignment{
-			Kind: scopedaccess.KindScopedRoleAssignment,
+			Kind:    scopedaccess.KindScopedRoleAssignment,
+			SubKind: scopedaccess.SubKindDynamic,
 			Metadata: &headerv1.Metadata{
 				Name: uuid.New().String(),
 			},
@@ -770,7 +773,8 @@ func TestUnscopedBasics(t *testing.T) {
 
 	// verify that admin can read the assignment
 	rasp, err := srvAlice.GetScopedRoleAssignment(ctx, &scopedaccessv1.GetScopedRoleAssignmentRequest{
-		Name: acrsp.GetAssignment().GetMetadata().GetName(),
+		Name:    acrsp.GetAssignment().GetMetadata().GetName(),
+		SubKind: acrsp.GetAssignment().GetSubKind(),
 	})
 	require.NoError(t, err)
 	require.Equal(t, acrsp.GetAssignment().GetMetadata().GetName(), rasp.GetAssignment().GetMetadata().GetName())
@@ -785,7 +789,8 @@ func TestUnscopedBasics(t *testing.T) {
 	// verify that auditor cannot create an assignment
 	_, err = srvBob.CreateScopedRoleAssignment(ctx, &scopedaccessv1.CreateScopedRoleAssignmentRequest{
 		Assignment: &scopedaccessv1.ScopedRoleAssignment{
-			Kind: scopedaccess.KindScopedRoleAssignment,
+			Kind:    scopedaccess.KindScopedRoleAssignment,
+			SubKind: scopedaccess.SubKindDynamic,
 			Metadata: &headerv1.Metadata{
 				Name: uuid.New().String(),
 			},
@@ -807,7 +812,8 @@ func TestUnscopedBasics(t *testing.T) {
 
 	// verify that auditor can read the admin-created assignment
 	rasp, err = srvBob.GetScopedRoleAssignment(ctx, &scopedaccessv1.GetScopedRoleAssignmentRequest{
-		Name: acrsp.GetAssignment().GetMetadata().GetName(),
+		Name:    acrsp.GetAssignment().GetMetadata().GetName(),
+		SubKind: acrsp.GetAssignment().GetSubKind(),
 	})
 	require.NoError(t, err)
 	require.Equal(t, acrsp.GetAssignment().GetMetadata().GetName(), rasp.GetAssignment().GetMetadata().GetName())
@@ -855,14 +861,16 @@ func TestUnscopedBasics(t *testing.T) {
 
 	// verify that auditor cannot delete assignments
 	_, err = srvBob.DeleteScopedRoleAssignment(ctx, &scopedaccessv1.DeleteScopedRoleAssignmentRequest{
-		Name: acrsp.GetAssignment().GetMetadata().GetName(),
+		Name:    acrsp.GetAssignment().GetMetadata().GetName(),
+		SubKind: acrsp.GetAssignment().GetSubKind(),
 	})
 	require.Error(t, err)
 	require.True(t, trace.IsAccessDenied(err), "expected access denied error, got: %v", err)
 
 	// verify that admin can delete assignments
 	_, err = srvAlice.DeleteScopedRoleAssignment(ctx, &scopedaccessv1.DeleteScopedRoleAssignmentRequest{
-		Name: acrsp.GetAssignment().GetMetadata().GetName(),
+		Name:    acrsp.GetAssignment().GetMetadata().GetName(),
+		SubKind: acrsp.GetAssignment().GetSubKind(),
 	})
 	require.NoError(t, err)
 

lib/auth/tls_test.go

@@ -1516,6 +1516,7 @@ func TestAuthPreferenceSettings_ScopedIdentity(t *testing.T) {
 	_, err = scopedSvc.CreateScopedRoleAssignment(ctx, &scopedaccessv1.CreateScopedRoleAssignmentRequest{
 		Assignment: &scopedaccessv1.ScopedRoleAssignment{
 			Kind:    scopedaccess.KindScopedRoleAssignment,
+			SubKind: scopedaccess.SubKindDynamic,
 			Version: types.V1,
 			Metadata: &headerv1.Metadata{
 				Name: uuid.NewString(),
@@ -2597,6 +2598,7 @@ func TestGetCertAuthority_ScopedIdentity(t *testing.T) {
 	_, err = scopedSvc.CreateScopedRoleAssignment(ctx, &scopedaccessv1.CreateScopedRoleAssignmentRequest{
 		Assignment: &scopedaccessv1.ScopedRoleAssignment{
 			Kind:    scopedaccess.KindScopedRoleAssignment,
+			SubKind: scopedaccess.SubKindDynamic,
 			Version: types.V1,
 			Metadata: &headerv1.Metadata{
 				Name: uuid.NewString(),

lib/client/api_login_test.go

@@ -775,7 +775,8 @@ func createAndAssignScopedRoles(t *testing.T, ctx context.Context, authServer *a
 	for _, role := range scopedRoles {
 		_, err = authServer.ScopedAccess().CreateScopedRoleAssignment(ctx, &scopedaccessv1.CreateScopedRoleAssignmentRequest{
 			Assignment: &scopedaccessv1.ScopedRoleAssignment{
-				Kind: scopedaccess.KindScopedRoleAssignment,
+				Kind:    scopedaccess.KindScopedRoleAssignment,
+				SubKind: scopedaccess.SubKindDynamic,
 				Metadata: &headerv1.Metadata{
 					Name: uuid.NewString(),
 				},

lib/scopes/access/access.go

@@ -45,6 +45,12 @@ const (
 	// KindScopedToken is the kind of a scoped token resource.
 	KindScopedToken = "scoped_token"
 
+	// SubKindDynamic is the sub kind of a scoped role assignment created via the API.
+	SubKindDynamic = "dynamic"
+
+	// SubKindMaterialized is the sub kind of a scoped role assignment that has been materialized.
+	SubKindMaterialized = "materialized"
+
 	// maxAssignableScopes is the maximum number of assignable scopes that a given scoped role resource may contain. Note that
 	// unlike MaxRolesPerAssignment, this is a fairly arbitrary limit and there isn't a strong reason to keep it low other than
 	// to avoid excess resource size and to keep our options open for the future.
@@ -321,6 +327,14 @@ func StrongValidateAssignment(assignment *scopedaccessv1.ScopedRoleAssignment) e
 		return trace.Wrap(err)
 	}
 
+	switch assignment.GetSubKind() {
+	case SubKindDynamic, SubKindMaterialized:
+	case "":
+		return trace.BadParameter("scoped role assignment %q has empty sub_kind", assignment.GetMetadata().GetName())
+	default:
+		return trace.BadParameter("scoped role assignment %q has invalid sub_kind %q", assignment.GetMetadata().GetName(), assignment.GetSubKind())
+	}
+
 	if _, err := uuid.Parse(assignment.GetMetadata().GetName()); err != nil {
 		return trace.BadParameter("scoped role assignment %q has invalid name (must be uuid): %v", assignment.GetMetadata().GetName(), err)
 	}
@@ -371,10 +385,6 @@ func commonValidateAssignment(assignment *scopedaccessv1.ScopedRoleAssignment) e
 		return trace.BadParameter("scoped role assignment %q has invalid kind %q, expected %q", assignment.GetMetadata().GetName(), assignment.GetKind(), KindScopedRoleAssignment)
 	}
 
-	if assignment.GetSubKind() != "" {
-		return trace.BadParameter("scoped role assignment %q has unknown sub_kind %q", assignment.GetMetadata().GetName(), assignment.GetSubKind())
-	}
-
 	if assignment.GetVersion() == "" {
 		return trace.BadParameter("scoped role assignment %q is missing version", assignment.GetMetadata().GetName())
 	}

lib/scopes/access/access_test.go

@@ -331,7 +331,8 @@ func TestValidateAsssignment(t *testing.T) {
 		{
 			name: "basic",
 			assignment: &scopedaccessv1.ScopedRoleAssignment{
-				Kind: KindScopedRoleAssignment,
+				Kind:    KindScopedRoleAssignment,
+				SubKind: SubKindDynamic,
 				Metadata: &headerv1.Metadata{
 					Name: uuid.New().String(),
 				},
@@ -371,7 +372,7 @@ func TestValidateAsssignment(t *testing.T) {
 				Version: types.V1,
 			},
 			strongOk: false,
-			weakOk:   false,
+			weakOk:   true,
 		},
 		{
 			name: "missing name",
@@ -415,12 +416,35 @@ func TestValidateAsssignment(t *testing.T) {
 			weakOk:   false,
 		},
 		{
-			name: "missing scope",
+			name: "missing sub_kind",
 			assignment: &scopedaccessv1.ScopedRoleAssignment{
 				Kind: KindScopedRoleAssignment,
 				Metadata: &headerv1.Metadata{
 					Name: uuid.New().String(),
 				},
+				Scope: "/",
+				Spec: &scopedaccessv1.ScopedRoleAssignmentSpec{
+					User: "alice",
+					Assignments: []*scopedaccessv1.Assignment{
+						{
+							Role:  "test",
+							Scope: "/foo",
+						},
+					},
+				},
+				Version: types.V1,
+			},
+			strongOk: false,
+			weakOk:   true,
+		},
+		{
+			name: "missing scope",
+			assignment: &scopedaccessv1.ScopedRoleAssignment{
+				Kind:    KindScopedRoleAssignment,
+				SubKind: SubKindDynamic,
+				Metadata: &headerv1.Metadata{
+					Name: uuid.New().String(),
+				},
 				Spec: &scopedaccessv1.ScopedRoleAssignmentSpec{
 					User: "alice",
 					Assignments: []*scopedaccessv1.Assignment{
@@ -438,7 +462,8 @@ func TestValidateAsssignment(t *testing.T) {
 		{
 			name: "missing version",
 			assignment: &scopedaccessv1.ScopedRoleAssignment{
-				Kind: KindScopedRoleAssignment,
+				Kind:    KindScopedRoleAssignment,
+				SubKind: SubKindDynamic,
 				Metadata: &headerv1.Metadata{
 					Name: uuid.New().String(),
 				},
@@ -459,7 +484,8 @@ func TestValidateAsssignment(t *testing.T) {
 		{
 			name: "malformed name",
 			assignment: &scopedaccessv1.ScopedRoleAssignment{
-				Kind: KindScopedRoleAssignment,
+				Kind:    KindScopedRoleAssignment,
+				SubKind: SubKindDynamic,
 				Metadata: &headerv1.Metadata{
 					Name: "not-a-uuid",
 				},
@@ -503,7 +529,8 @@ func TestValidateAsssignment(t *testing.T) {
 		{
 			name: "slightly malformed scope",
 			assignment: &scopedaccessv1.ScopedRoleAssignment{
-				Kind: KindScopedRoleAssignment,
+				Kind:    KindScopedRoleAssignment,
+				SubKind: SubKindDynamic,
 				Metadata: &headerv1.Metadata{
 					Name: uuid.New().String(),
 				},
@@ -525,7 +552,8 @@ func TestValidateAsssignment(t *testing.T) {
 		{
 			name: "significantly malformed scope",
 			assignment: &scopedaccessv1.ScopedRoleAssignment{
-				Kind: KindScopedRoleAssignment,
+				Kind:    KindScopedRoleAssignment,
+				SubKind: SubKindDynamic,
 				Metadata: &headerv1.Metadata{
 					Name: uuid.New().String(),
 				},
@@ -547,7 +575,8 @@ func TestValidateAsssignment(t *testing.T) {
 		{
 			name: "impermissable assigned scope",
 			assignment: &scopedaccessv1.ScopedRoleAssignment{
-				Kind: KindScopedRoleAssignment,
+				Kind:    KindScopedRoleAssignment,
+				SubKind: SubKindDynamic,
 				Metadata: &headerv1.Metadata{
 					Name: uuid.New().String(),
 				},
@@ -569,7 +598,8 @@ func TestValidateAsssignment(t *testing.T) {
 		{
 			name: "malformed assigned scope",
 			assignment: &scopedaccessv1.ScopedRoleAssignment{
-				Kind: KindScopedRoleAssignment,
+				Kind:    KindScopedRoleAssignment,
+				SubKind: SubKindDynamic,
 				Metadata: &headerv1.Metadata{
 					Name: uuid.New().String(),
 				},
@@ -591,7 +621,8 @@ func TestValidateAsssignment(t *testing.T) {
 		{
 			name: "basic",
 			assignment: &scopedaccessv1.ScopedRoleAssignment{
-				Kind: KindScopedRoleAssignment,
+				Kind:    KindScopedRoleAssignment,
+				SubKind: SubKindDynamic,
 				Metadata: &headerv1.Metadata{
 					Name: uuid.New().String(),
 				},

lib/scopes/cache/access/access.go

@@ -350,7 +350,7 @@ func processEvent(ctx context.Context, state state, event types.Event) error {
 		case scopedaccess.KindScopedRole:
 			state.roles.Delete(event.Resource.GetName())
 		case scopedaccess.KindScopedRoleAssignment:
-			state.assignments.Delete(event.Resource.GetName())
+			state.assignments.Delete(event.Resource.GetName(), event.Resource.GetSubKind())
 		default:
 			return trace.BadParameter("unexpected resource kind %q in event delete event", event.Resource.GetKind())
 		}

lib/scopes/cache/access/access_test.go

@@ -195,7 +195,8 @@ func TestScopedAccessCacheReplication(t *testing.T) {
 	// test that cache can handle partial deletes for assignments
 	for _, name := range expectedAssignmentNames[0:10] {
 		_, err := service.DeleteScopedRoleAssignment(ctx, &scopedaccessv1.DeleteScopedRoleAssignmentRequest{
-			Name: name,
+			Name:    name,
+			SubKind: scopedaccess.SubKindDynamic,
 		})
 		require.NoError(t, err)
 	}
@@ -216,7 +217,8 @@ func TestScopedAccessCacheReplication(t *testing.T) {
 	// test that cache can handle delete of all assignments
 	for _, name := range expectedAssignmentNames[10:] {
 		_, err := service.DeleteScopedRoleAssignment(ctx, &scopedaccessv1.DeleteScopedRoleAssignmentRequest{
-			Name: name,
+			Name:    name,
+			SubKind: scopedaccess.SubKindDynamic,
 		})
 		require.NoError(t, err)
 	}
@@ -409,7 +411,8 @@ func newScopedRole(name string) *scopedaccessv1.ScopedRole {
 
 func newScopedRoleAssignment(roleName string) *scopedaccessv1.ScopedRoleAssignment {
 	return &scopedaccessv1.ScopedRoleAssignment{
-		Kind: scopedaccess.KindScopedRoleAssignment,
+		Kind:    scopedaccess.KindScopedRoleAssignment,
+		SubKind: scopedaccess.SubKindDynamic,
 		Metadata: &headerv1.Metadata{
 			Name: uuid.New().String(),
 		},

lib/scopes/cache/assignments/assignment_cache.go

@@ -73,7 +73,10 @@ func NewAssignmentCache(cfg AssignmentCacheConfig) *AssignmentCache {
 				return assignment.GetScope()
 			},
 			Key: func(assignment *scopedaccessv1.ScopedRoleAssignment) string {
-				return assignment.GetMetadata().GetName()
+				return assignmentKey{
+					name:    assignment.GetMetadata().GetName(),
+					subKind: assignment.GetSubKind(),
+				}.String()
 			},
 			Clone: proto.CloneOf[*scopedaccessv1.ScopedRoleAssignment],
 		}),
@@ -87,7 +90,10 @@ func (c *AssignmentCache) GetScopedRoleAssignment(ctx context.Context, req *scop
 		return nil, trace.BadParameter("missing scoped role assignment name in get request")
 	}
 
-	assignment, ok := c.cache.Get(req.GetName())
+	assignment, ok := c.cache.Get(assignmentKey{
+		name:    req.GetName(),
+		subKind: req.GetSubKind(),
+	}.String())
 	if !ok {
 		return nil, trace.NotFound("scoped role assignment %q not found", req.GetName())
 	}
@@ -180,7 +186,10 @@ Outer:
 			if len(out) == pageSize {
 				nextCursor = cache.Cursor[string]{
 					Scope: scope.Scope(),
-					Key:   assignment.GetMetadata().GetName(),
+					Key: assignmentKey{
+						name:    assignment.GetMetadata().GetName(),
+						subKind: assignment.GetSubKind(),
+					}.String(),
 				}
 				break Outer
 			}
@@ -212,7 +221,22 @@ func (c *AssignmentCache) Put(assignment *scopedaccessv1.ScopedRoleAssignment) e
 	return nil
 }
 
-// Del removes an assignment from the cache by name.
-func (c *AssignmentCache) Delete(name string) {
-	c.cache.Del(name)
+// Delete removes an assignment from the cache by name.
+func (c *AssignmentCache) Delete(name, subKind string) {
+	c.cache.Del(assignmentKey{
+		name:    name,
+		subKind: subKind,
+	}.String())
+}
+
+type assignmentKey struct {
+	name    string
+	subKind string
+}
+
+func (k assignmentKey) String() string {
+	if k.subKind == "" {
+		return k.name
+	}
+	return k.name + "/" + k.subKind
 }

lib/scopes/cache/assignments/assignment_cache_test.go

@@ -38,7 +38,8 @@ func TestListScopedRoleAssignmentsScenarios(t *testing.T) {
 
 	assignments := []*scopedaccessv1.ScopedRoleAssignment{
 		{
-			Kind: scopedaccess.KindScopedRoleAssignment,
+			Kind:    scopedaccess.KindScopedRoleAssignment,
+			SubKind: scopedaccess.SubKindDynamic,
 			Metadata: &headerpb.Metadata{
 				Name: "alice-01",
 			},
@@ -59,7 +60,8 @@ func TestListScopedRoleAssignmentsScenarios(t *testing.T) {
 			Version: types.V1,
 		},
 		{
-			Kind: scopedaccess.KindScopedRoleAssignment,
+			Kind:    scopedaccess.KindScopedRoleAssignment,
+			SubKind: scopedaccess.SubKindDynamic,
 			Metadata: &headerpb.Metadata{
 				Name: "alice-02",
 			},
@@ -80,7 +82,8 @@ func TestListScopedRoleAssignmentsScenarios(t *testing.T) {
 			Version: types.V1,
 		},
 		{
-			Kind: scopedaccess.KindScopedRoleAssignment,
+			Kind:    scopedaccess.KindScopedRoleAssignment,
+			SubKind: scopedaccess.SubKindDynamic,
 			Metadata: &headerpb.Metadata{
 				Name: "bob-01",
 			},
@@ -101,7 +104,8 @@ func TestListScopedRoleAssignmentsScenarios(t *testing.T) {
 			Version: types.V1,
 		},
 		{
-			Kind: scopedaccess.KindScopedRoleAssignment,
+			Kind:    scopedaccess.KindScopedRoleAssignment,
+			SubKind: scopedaccess.SubKindDynamic,
 			Metadata: &headerpb.Metadata{
 				Name: "bob-02",
 			},
@@ -122,7 +126,8 @@ func TestListScopedRoleAssignmentsScenarios(t *testing.T) {
 			Version: types.V1,
 		},
 		{
-			Kind: scopedaccess.KindScopedRoleAssignment,
+			Kind:    scopedaccess.KindScopedRoleAssignment,
+			SubKind: scopedaccess.SubKindDynamic,
 			Metadata: &headerpb.Metadata{
 				Name: "alice-03",
 			},
@@ -143,7 +148,8 @@ func TestListScopedRoleAssignmentsScenarios(t *testing.T) {
 			Version: types.V1,
 		},
 		{
-			Kind: scopedaccess.KindScopedRoleAssignment,
+			Kind:    scopedaccess.KindScopedRoleAssignment,
+			SubKind: scopedaccess.SubKindDynamic,
 			Metadata: &headerpb.Metadata{
 				Name: "bob-03",
 			},
@@ -164,7 +170,8 @@ func TestListScopedRoleAssignmentsScenarios(t *testing.T) {
 			Version: types.V1,
 		},
 		{
-			Kind: scopedaccess.KindScopedRoleAssignment,
+			Kind:    scopedaccess.KindScopedRoleAssignment,
+			SubKind: scopedaccess.SubKindDynamic,
 			Metadata: &headerpb.Metadata{
 				Name: "carol-01",
 			},
@@ -189,7 +196,8 @@ func TestListScopedRoleAssignmentsScenarios(t *testing.T) {
 	cache := NewAssignmentCache(AssignmentCacheConfig{})
 	for _, assignment := range assignments {
 		_, err := cache.GetScopedRoleAssignment(t.Context(), &scopedaccessv1.GetScopedRoleAssignmentRequest{
-			Name: assignment.GetMetadata().GetName(),
+			Name:    assignment.GetMetadata().GetName(),
+			SubKind: assignment.GetSubKind(),
 		})
 		require.Error(t, err)
 		require.True(t, trace.IsNotFound(err), "expected NotFound error, got %v", err)
@@ -197,7 +205,8 @@ func TestListScopedRoleAssignmentsScenarios(t *testing.T) {
 		cache.Put(assignment)
 
 		rsp, err := cache.GetScopedRoleAssignment(t.Context(), &scopedaccessv1.GetScopedRoleAssignmentRequest{
-			Name: assignment.GetMetadata().GetName(),
+			Name:    assignment.GetMetadata().GetName(),
+			SubKind: assignment.GetSubKind(),
 		})
 		require.NoError(t, err)
 		require.NotNil(t, rsp.GetAssignment())
@@ -209,7 +218,7 @@ func TestListScopedRoleAssignmentsScenarios(t *testing.T) {
 			Mode:  scopespb.Mode_MODE_RESOURCES_SUBJECT_TO_SCOPE,
 			Scope: "/aa",
 		},
-		PageToken: "v1:bob-02@/aa",
+		PageToken: "v1:bob-02/dynamic@/aa",
 	})
 	require.NoError(t, err)
 	require.Empty(t, rsp.NextPageToken)
@@ -221,7 +230,7 @@ func TestListScopedRoleAssignmentsScenarios(t *testing.T) {
 			Mode:  scopespb.Mode_MODE_RESOURCES_SUBJECT_TO_SCOPE,
 			Scope: "/aa",
 		},
-		PageToken: "v1:bob-01@/",
+		PageToken: "v1:bob-01/dynamic@/",
 	})
 	require.NoError(t, err)
 	require.Empty(t, rsp.NextPageToken)
@@ -233,7 +242,7 @@ func TestListScopedRoleAssignmentsScenarios(t *testing.T) {
 			Mode:  scopespb.Mode_MODE_RESOURCES_SUBJECT_TO_SCOPE,
 			Scope: "/aa",
 		},
-		PageToken: "v1:carol-01@/bb",
+		PageToken: "v1:carol-01/dynamic@/bb",
 	})
 	require.NoError(t, err)
 	require.Empty(t, rsp.NextPageToken)
@@ -245,7 +254,7 @@ func TestListScopedRoleAssignmentsScenarios(t *testing.T) {
 			Mode:  scopespb.Mode_MODE_POLICIES_APPLICABLE_TO_SCOPE,
 			Scope: "/aa",
 		},
-		PageToken: "v1:bob-01@/",
+		PageToken: "v1:bob-01/dynamic@/",
 	})
 	require.NoError(t, err)
 	require.Empty(t, rsp.NextPageToken)
@@ -258,7 +267,7 @@ func TestListScopedRoleAssignmentsScenarios(t *testing.T) {
 			Mode:  scopespb.Mode_MODE_POLICIES_APPLICABLE_TO_SCOPE,
 			Scope: "/aa",
 		},
-		PageToken: "v1:bob-03@/aa/bb",
+		PageToken: "v1:bob-03/dynamic@/aa/bb",
 	})
 	require.NoError(t, err)
 	require.Empty(t, rsp.NextPageToken)
@@ -271,7 +280,7 @@ func TestListScopedRoleAssignmentsScenarios(t *testing.T) {
 			Mode:  scopespb.Mode_MODE_POLICIES_APPLICABLE_TO_SCOPE,
 			Scope: "/aa",
 		},
-		PageToken: "v1:carol-01@/bb",
+		PageToken: "v1:carol-01/dynamic@/bb",
 	})
 	require.NoError(t, err)
 	require.Empty(t, rsp.NextPageToken)
@@ -283,7 +292,7 @@ func TestListScopedRoleAssignmentsScenarios(t *testing.T) {
 			Mode:  scopespb.Mode_MODE_RESOURCES_SUBJECT_TO_SCOPE,
 			Scope: "/aa",
 		},
-		PageToken: "v2:bob-02@/aa",
+		PageToken: "v2:bob-02/dynamic@/aa",
 	})
 	require.Error(t, err)
 	require.True(t, trace.IsBadParameter(err), "expected BadParameter error, got %v", err)
@@ -295,7 +304,8 @@ func TestListScopedRoleAssignmentsBasics(t *testing.T) {
 
 	assignments := []*scopedaccessv1.ScopedRoleAssignment{
 		{
-			Kind: scopedaccess.KindScopedRoleAssignment,
+			Kind:    scopedaccess.KindScopedRoleAssignment,
+			SubKind: scopedaccess.SubKindDynamic,
 			Metadata: &headerpb.Metadata{
 				Name: "alice-01",
 			},
@@ -316,7 +326,8 @@ func TestListScopedRoleAssignmentsBasics(t *testing.T) {
 			Version: types.V1,
 		},
 		{
-			Kind: scopedaccess.KindScopedRoleAssignment,
+			Kind:    scopedaccess.KindScopedRoleAssignment,
+			SubKind: scopedaccess.SubKindDynamic,
 			Metadata: &headerpb.Metadata{
 				Name: "alice-02",
 			},
@@ -337,7 +348,8 @@ func TestListScopedRoleAssignmentsBasics(t *testing.T) {
 			Version: types.V1,
 		},
 		{
-			Kind: scopedaccess.KindScopedRoleAssignment,
+			Kind:    scopedaccess.KindScopedRoleAssignment,
+			SubKind: scopedaccess.SubKindDynamic,
 			Metadata: &headerpb.Metadata{
 				Name: "bob-01",
 			},
@@ -358,7 +370,8 @@ func TestListScopedRoleAssignmentsBasics(t *testing.T) {
 			Version: types.V1,
 		},
 		{
-			Kind: scopedaccess.KindScopedRoleAssignment,
+			Kind:    scopedaccess.KindScopedRoleAssignment,
+			SubKind: scopedaccess.SubKindDynamic,
 			Metadata: &headerpb.Metadata{
 				Name: "bob-02",
 			},
@@ -379,7 +392,8 @@ func TestListScopedRoleAssignmentsBasics(t *testing.T) {
 			Version: types.V1,
 		},
 		{
-			Kind: scopedaccess.KindScopedRoleAssignment,
+			Kind:    scopedaccess.KindScopedRoleAssignment,
+			SubKind: scopedaccess.SubKindDynamic,
 			Metadata: &headerpb.Metadata{
 				Name: "carol-01",
 			},
@@ -598,7 +612,8 @@ func TestListScopedRoleAssignmentsBasics(t *testing.T) {
 	cache := NewAssignmentCache(AssignmentCacheConfig{})
 	for _, assignment := range assignments {
 		_, err := cache.GetScopedRoleAssignment(t.Context(), &scopedaccessv1.GetScopedRoleAssignmentRequest{
-			Name: assignment.GetMetadata().GetName(),
+			Name:    assignment.GetMetadata().GetName(),
+			SubKind: assignment.GetSubKind(),
 		})
 		require.Error(t, err)
 		require.True(t, trace.IsNotFound(err), "expected NotFound error, got %v", err)
@@ -606,7 +621,8 @@ func TestListScopedRoleAssignmentsBasics(t *testing.T) {
 		cache.Put(assignment)
 
 		rsp, err := cache.GetScopedRoleAssignment(t.Context(), &scopedaccessv1.GetScopedRoleAssignmentRequest{
-			Name: assignment.GetMetadata().GetName(),
+			Name:    assignment.GetMetadata().GetName(),
+			SubKind: assignment.GetSubKind(),
 		})
 		require.NoError(t, err)
 		require.NotNil(t, rsp.GetAssignment())
@@ -637,6 +653,87 @@ func TestListScopedRoleAssignmentsBasics(t *testing.T) {
 	}
 }
 
+// TestScopedRoleAssignmentSubKinds asserts that assignments with different
+// subkinds are properly fetched and listed with pagination, even if their
+// names collide.
+func TestScopedRoleAssignmentSubKinds(t *testing.T) {
+	t.Parallel()
+
+	cache := NewAssignmentCache(AssignmentCacheConfig{})
+
+	makeAssignment := func(name, subKind, user string) *scopedaccessv1.ScopedRoleAssignment {
+		return &scopedaccessv1.ScopedRoleAssignment{
+			Kind:    scopedaccess.KindScopedRoleAssignment,
+			SubKind: subKind,
+			Metadata: &headerpb.Metadata{
+				Name: name,
+			},
+			Scope: "/",
+			Spec: &scopedaccessv1.ScopedRoleAssignmentSpec{
+				User: user,
+				Assignments: []*scopedaccessv1.Assignment{{
+					Role:  "role-01",
+					Scope: "/foo",
+				}},
+			},
+			Version: types.V1,
+		}
+	}
+
+	// Populate the cache with a dynamic and a materialized assignment with the same name.
+	dynamic := makeAssignment("shared-name", scopedaccess.SubKindDynamic, "alice")
+	materialized := makeAssignment("shared-name", scopedaccess.SubKindMaterialized, "bob")
+	require.NoError(t, cache.Put(dynamic))
+	require.NoError(t, cache.Put(materialized))
+
+	// Make sure getting each assignment by (name, subkind) returns the right one.
+	for _, tt := range []*scopedaccessv1.ScopedRoleAssignment{dynamic, materialized} {
+		rsp, err := cache.GetScopedRoleAssignment(t.Context(), &scopedaccessv1.GetScopedRoleAssignmentRequest{
+			Name:    tt.GetMetadata().GetName(),
+			SubKind: tt.GetSubKind(),
+		})
+		require.NoError(t, err)
+		require.Equal(t, tt.GetSpec().GetUser(), rsp.GetAssignment().GetSpec().GetUser())
+		require.Equal(t, tt.GetSubKind(), rsp.GetAssignment().GetSubKind())
+	}
+
+	// Trying to get an assignment without specifying a subkind returns NotFound.
+	_, err := cache.GetScopedRoleAssignment(t.Context(), &scopedaccessv1.GetScopedRoleAssignmentRequest{
+		Name: "shared-name",
+	})
+	require.Error(t, err)
+	require.True(t, trace.IsNotFound(err), "expected NotFound error, got %v", err)
+
+	// Make sure paging across assignments does not skip or duplicate
+	// assignments with matching names but different subkinds.
+	var got []string
+	pageToken := ""
+	for {
+		rsp, err := cache.ListScopedRoleAssignments(t.Context(), &scopedaccessv1.ListScopedRoleAssignmentsRequest{
+			PageSize:  1,
+			PageToken: pageToken,
+		})
+		require.NoError(t, err)
+		require.Len(t, rsp.GetAssignments(), 1)
+
+		assignment := rsp.GetAssignments()[0]
+		got = append(got, assignment.GetMetadata().GetName()+"/"+assignment.GetSubKind())
+
+		if rsp.GetNextPageToken() == "" {
+			break
+		}
+		pageToken = rsp.GetNextPageToken()
+	}
+
+	require.ElementsMatch(t,
+		[]string{
+			"shared-name/dynamic",
+			"shared-name/materialized",
+		},
+		got,
+	)
+}
+
 func collectAssignmentNames(assignments []*scopedaccessv1.ScopedRoleAssignment) []string {
 	var names []string
 	for _, assignment := range assignments {

lib/scopes/cache/assignments/pinning_test.go

@@ -41,7 +41,8 @@ func TestPopulatePinnedAssignmentsForUser(t *testing.T) {
 
 	assignments := []*scopedaccessv1.ScopedRoleAssignment{
 		{
-			Kind: scopedaccess.KindScopedRoleAssignment,
+			Kind:    scopedaccess.KindScopedRoleAssignment,
+			SubKind: scopedaccess.SubKindDynamic,
 			Metadata: &headerpb.Metadata{
 				Name: "alice-01",
 			},
@@ -62,7 +63,8 @@ func TestPopulatePinnedAssignmentsForUser(t *testing.T) {
 			Version: types.V1,
 		},
 		{
-			Kind: scopedaccess.KindScopedRoleAssignment,
+			Kind:    scopedaccess.KindScopedRoleAssignment,
+			SubKind: scopedaccess.SubKindDynamic,
 			Metadata: &headerpb.Metadata{
 				Name: "alice-02",
 			},
@@ -83,7 +85,8 @@ func TestPopulatePinnedAssignmentsForUser(t *testing.T) {
 			Version: types.V1,
 		},
 		{
-			Kind: scopedaccess.KindScopedRoleAssignment,
+			Kind:    scopedaccess.KindScopedRoleAssignment,
+			SubKind: scopedaccess.SubKindDynamic,
 			Metadata: &headerpb.Metadata{
 				Name: "bob-01",
 			},
@@ -104,7 +107,8 @@ func TestPopulatePinnedAssignmentsForUser(t *testing.T) {
 			Version: types.V1,
 		},
 		{
-			Kind: scopedaccess.KindScopedRoleAssignment,
+			Kind:    scopedaccess.KindScopedRoleAssignment,
+			SubKind: scopedaccess.SubKindDynamic,
 			Metadata: &headerpb.Metadata{
 				Name: "bob-02",
 			},
@@ -125,7 +129,8 @@ func TestPopulatePinnedAssignmentsForUser(t *testing.T) {
 			Version: types.V1,
 		},
 		{
-			Kind: scopedaccess.KindScopedRoleAssignment,
+			Kind:    scopedaccess.KindScopedRoleAssignment,
+			SubKind: scopedaccess.SubKindDynamic,
 			Metadata: &headerpb.Metadata{
 				Name: "alice-03",
 			},
@@ -146,7 +151,8 @@ func TestPopulatePinnedAssignmentsForUser(t *testing.T) {
 			Version: types.V1,
 		},
 		{
-			Kind: scopedaccess.KindScopedRoleAssignment,
+			Kind:    scopedaccess.KindScopedRoleAssignment,
+			SubKind: scopedaccess.SubKindDynamic,
 			Metadata: &headerpb.Metadata{
 				Name: "bob-03",
 			},
@@ -167,7 +173,8 @@ func TestPopulatePinnedAssignmentsForUser(t *testing.T) {
 			Version: types.V1,
 		},
 		{
-			Kind: scopedaccess.KindScopedRoleAssignment,
+			Kind:    scopedaccess.KindScopedRoleAssignment,
+			SubKind: scopedaccess.SubKindDynamic,
 			Metadata: &headerpb.Metadata{
 				Name: "carol-01",
 			},
@@ -192,7 +199,8 @@ func TestPopulatePinnedAssignmentsForUser(t *testing.T) {
 	cache := NewAssignmentCache(AssignmentCacheConfig{})
 	for _, assignment := range assignments {
 		_, err := cache.GetScopedRoleAssignment(t.Context(), &scopedaccessv1.GetScopedRoleAssignmentRequest{
-			Name: assignment.GetMetadata().GetName(),
+			Name:    assignment.GetMetadata().GetName(),
+			SubKind: assignment.GetSubKind(),
 		})
 		require.Error(t, err)
 		require.True(t, trace.IsNotFound(err), "expected NotFound error, got %v", err)
@@ -200,7 +208,8 @@ func TestPopulatePinnedAssignmentsForUser(t *testing.T) {
 		cache.Put(assignment)
 
 		rsp, err := cache.GetScopedRoleAssignment(t.Context(), &scopedaccessv1.GetScopedRoleAssignmentRequest{
-			Name: assignment.GetMetadata().GetName(),
+			Name:    assignment.GetMetadata().GetName(),
+			SubKind: assignment.GetSubKind(),
 		})
 		require.NoError(t, err)
 		require.NotNil(t, rsp.GetAssignment())
@@ -284,7 +293,8 @@ func TestAssignmentTreePruning(t *testing.T) {
 	// must have a mix of resource scopes to provide a natural pruning boundary.
 	assignments := []*scopedaccessv1.ScopedRoleAssignment{
 		{
-			Kind: scopedaccess.KindScopedRoleAssignment,
+			Kind:    scopedaccess.KindScopedRoleAssignment,
+			SubKind: scopedaccess.SubKindDynamic,
 			Metadata: &headerpb.Metadata{
 				Name: "alice-root",
 			},
@@ -298,7 +308,8 @@ func TestAssignmentTreePruning(t *testing.T) {
 			Version: types.V1,
 		},
 		{
-			Kind: scopedaccess.KindScopedRoleAssignment,
+			Kind:    scopedaccess.KindScopedRoleAssignment,
+			SubKind: scopedaccess.SubKindDynamic,
 			Metadata: &headerpb.Metadata{
 				Name: "alice-staging",
 			},
@@ -312,7 +323,8 @@ func TestAssignmentTreePruning(t *testing.T) {
 			Version: types.V1,
 		},
 		{
-			Kind: scopedaccess.KindScopedRoleAssignment,
+			Kind:    scopedaccess.KindScopedRoleAssignment,
+			SubKind: scopedaccess.SubKindDynamic,
 			Metadata: &headerpb.Metadata{
 				Name: "alice-staging-west",
 			},

lib/scopes/utils/range_test.go

@@ -19,7 +19,6 @@
 package utils
 
 import (
-	"context"
 	"fmt"
 	"testing"
 
@@ -38,9 +37,7 @@ func TestRangeScopedRoles(t *testing.T) {
 	const roleCount = 503
 
 	t.Parallel()
-
-	ctx, cancel := context.WithCancel(context.Background())
-	defer cancel()
+	ctx := t.Context()
 
 	upstream := roles.NewRoleCache()
 
@@ -79,9 +76,7 @@ func TestRangeScopedRoleAssignments(t *testing.T) {
 	const assignmentCount = 503
 
 	t.Parallel()
-
-	ctx, cancel := context.WithCancel(context.Background())
-	defer cancel()
+	ctx := t.Context()
 
 	upstream := assignments.NewAssignmentCache(assignments.AssignmentCacheConfig{})
 
@@ -90,7 +85,8 @@ func TestRangeScopedRoleAssignments(t *testing.T) {
 	for range assignmentCount {
 		name := uuid.New().String()
 		err := upstream.Put(&scopedaccessv1.ScopedRoleAssignment{
-			Kind: scopedaccess.KindScopedRoleAssignment,
+			Kind:    scopedaccess.KindScopedRoleAssignment,
+			SubKind: scopedaccess.SubKindDynamic,
 			Metadata: &headerv1.Metadata{
 				Name: name,
 			},

lib/services/local/events.go

@@ -1113,12 +1113,30 @@ type scopedRoleAssignmentParser struct {
 func (p *scopedRoleAssignmentParser) parse(event backend.Event) (types.Resource, error) {
 	switch event.Type {
 	case types.OpDelete:
-		name := strings.TrimPrefix(event.Item.Key.TrimPrefix(scopedRoleAssignmentWatchPrefix()).String(), backend.SeparatorString)
-		if name == "" || strings.Contains(name, "/") {
+		components := event.Item.Key.TrimPrefix(scopedRoleAssignmentWatchPrefix()).Components()
+		name := ""
+		subKind := ""
+		switch len(components) {
+		case 1:
+			name = components[0]
+		case 2:
+			name = components[0]
+			subKind = components[1]
+		default:
 			return nil, trace.NotFound("failed parsing %v", event.Item.Key.String())
 		}
+		if name == "" {
+			return nil, trace.NotFound("failed parsing %v", event.Item.Key.String())
+		}
+		if subKind == scopedaccess.SubKindMaterialized {
+			// Materialized assignments are filtered out from backend events in
+			// case a future version persists materialized assignments to the
+			// backend.
+			return nil, nil
+		}
 		return &types.ResourceHeader{
-			Kind: scopedaccess.KindScopedRoleAssignment,
+			Kind:    scopedaccess.KindScopedRoleAssignment,
+			SubKind: subKind,
 			Metadata: types.Metadata{
 				Name: name,
 			},
@@ -1128,6 +1146,12 @@ func (p *scopedRoleAssignmentParser) parse(event backend.Event) (types.Resource,
 		if err != nil {
 			return nil, trace.Wrap(err)
 		}
+		if assignment.GetSubKind() == scopedaccess.SubKindMaterialized {
+			// Materialized assignments are filtered out from backend events in
+			// case a future version persists materialized assignments to the
+			// backend.
+			return nil, nil
+		}
 		return types.Resource153ToLegacy(assignment), nil
 	default:
 		return nil, trace.BadParameter("event %v is not supported", event.Type)

lib/services/local/scoped_access.go

@@ -407,8 +407,15 @@ func (s *ScopedAccessService) GetScopedRoleAssignment(ctx context.Context, req *
 	if assignmentName == "" {
 		return nil, trace.BadParameter("missing scoped role assignment name in get request")
 	}
+	subKind := req.GetSubKind()
+	if subKind == scopedaccess.SubKindMaterialized {
+		return nil, trace.BadParameter(`reading scoped role assignments with sub_kind "materialized" from the backend is not supported`)
+	}
 
-	item, err := s.bk.Get(ctx, scopedRoleAssignmentKey(assignmentName))
+	item, err := s.bk.Get(ctx, scopedRoleAssignmentKey{
+		name:    assignmentName,
+		subKind: subKind,
+	}.Key())
 	if err != nil {
 		if trace.IsNotFound(err) {
 			return nil, trace.NotFound("scoped role assignment %q not found", assignmentName)
@@ -462,7 +469,7 @@ func (s *ScopedAccessService) ListScopedRoleAssignments(ctx context.Context, req
 // Returned assignments have had weak validation applied.
 func (s *ScopedAccessService) StreamScopedRoleAssignments(ctx context.Context) stream.Stream[*scopedaccessv1.ScopedRoleAssignment] {
 	return func(yield func(*scopedaccessv1.ScopedRoleAssignment, error) bool) {
-		startKey := scopedRoleAssignmentKey("")
+		startKey := scopedRoleAssignmentWatchPrefix()
 		params := backend.ItemsParams{
 			StartKey: startKey,
 			EndKey:   backend.RangeEnd(startKey),
@@ -482,6 +489,14 @@ func (s *ScopedAccessService) StreamScopedRoleAssignments(ctx context.Context) s
 				continue
 			}
 
+			if assignment.GetSubKind() == scopedaccess.SubKindMaterialized {
+				// Reading materialized assignments from the backend is not
+				// currently supported, we skip them in case materialized
+				// assignments are persisted to the backend in a future
+				// version.
+				continue
+			}
+
 			if err := scopedaccess.WeakValidateAssignment(assignment); err != nil {
 				// per-assignment errors are logged and skipped
 				s.logger.WarnContext(ctx, "skipping scoped role assignment due to validation error", "error", err, "key", logutils.StringerAttr(item.Key))
@@ -505,6 +520,12 @@ func (s *ScopedAccessService) CreateScopedRoleAssignment(ctx context.Context, re
 		return nil, trace.Wrap(err)
 	}
 
+	switch assignment.GetSubKind() {
+	case scopedaccess.SubKindDynamic:
+	default:
+		return nil, trace.BadParameter("creating scoped role assignments with sub_kind %q is not supported", assignment.GetSubKind())
+	}
+
 	// independently enforce the max number of roles per assignment limit here since not all validation
 	// may necessarily enforce it, but it is a hard-limit for the backend impl.
 	if len(assignment.GetSpec().GetAssignments()) > scopedaccess.MaxRolesPerAssignment {
@@ -605,12 +626,22 @@ func (s *ScopedAccessService) DeleteScopedRoleAssignment(ctx context.Context, re
 		return nil, trace.BadParameter("missing scoped role assignment name in delete request")
 	}
 
+	subKind := req.GetSubKind()
+	switch subKind {
+	case scopedaccess.SubKindDynamic, "":
+	case scopedaccess.SubKindMaterialized:
+		return nil, trace.BadParameter(`deleting scoped role assignments with sub_kind "materialized" is not supported`)
+	default:
+		return nil, trace.BadParameter("unhandled sub_kind %q in scoped role assignment delete request", subKind)
+	}
+
 	extant, err := s.GetScopedRoleAssignment(ctx, &scopedaccessv1.GetScopedRoleAssignmentRequest{
-		Name: assignmentName,
+		Name:    assignmentName,
+		SubKind: subKind,
 	})
 	if err != nil {
 		if trace.IsNotFound(err) {
-			return nil, trace.CompareFailed("scoped role assignment %q was concurrently delete", assignmentName)
+			return nil, trace.CompareFailed("scoped role assignment %q was concurrently deleted", assignmentName)
 		}
 		return nil, trace.Wrap(err)
 	}
@@ -649,7 +680,8 @@ func (s *ScopedAccessService) DeleteScopedRoleAssignment(ctx context.Context, re
 				// skip assignments related to other users
 				continue
 			}
-			if assignment.GetMetadata().GetName() == extant.Assignment.GetMetadata().GetName() {
+			if assignment.GetMetadata().GetName() == extant.Assignment.GetMetadata().GetName() &&
+				assignment.GetSubKind() == extant.Assignment.GetSubKind() {
 				// skip the assignment we're currently deleting
 				continue
 			}
@@ -667,7 +699,10 @@ func (s *ScopedAccessService) DeleteScopedRoleAssignment(ctx context.Context, re
 
 	condacts := []backend.ConditionalAction{
 		{
-			Key:       scopedRoleAssignmentKey(assignmentName),
+			Key: scopedRoleAssignmentKey{
+				name:    assignmentName,
+				subKind: extant.Assignment.GetSubKind(),
+			}.Key(),
 			Condition: backend.Revision(extant.Assignment.GetMetadata().GetRevision()),
 			Action:    backend.Delete(),
 		},
@@ -837,8 +872,17 @@ func scopedRoleWatchPrefix() backend.Key {
 	return backend.ExactKey(scopedRolePrefix, scopedRoleRoleComponent)
 }
 
-func scopedRoleAssignmentKey(assignmentID string) backend.Key {
-	return backend.NewKey(scopedRolePrefix, scopedRoleAssignmentComponent, assignmentID)
+type scopedRoleAssignmentKey struct {
+	name    string
+	subKind string
+}
+
+func (k scopedRoleAssignmentKey) Key() backend.Key {
+	if k.subKind == "" {
+		// Supports reading old scoped role assignments created without a subkind.
+		return backend.NewKey(scopedRolePrefix, scopedRoleAssignmentComponent, k.name)
+	}
+	return backend.NewKey(scopedRolePrefix, scopedRoleAssignmentComponent, k.name, k.subKind)
 }
 
 func scopedRoleAssignmentWatchPrefix() backend.Key {
@@ -926,13 +970,20 @@ func scopedRoleAssignmentToItem(assignment *scopedaccessv1.ScopedRoleAssignment)
 		return backend.Item{}, trace.BadParameter("scoped role assignments do not support expiration")
 	}
 
+	if assignment.GetSubKind() == "" {
+		return backend.Item{}, trace.BadParameter("scoped role assignments must have a sub_kind")
+	}
+
 	data, err := protojson.Marshal(assignment)
 	if err != nil {
 		return backend.Item{}, trace.Wrap(err)
 	}
 
 	return backend.Item{
-		Key:      scopedRoleAssignmentKey(assignment.GetMetadata().GetName()),
+		Key: scopedRoleAssignmentKey{
+			name:    assignment.GetMetadata().GetName(),
+			subKind: assignment.GetSubKind(),
+		}.Key(),
 		Value:    data,
 		Revision: assignment.GetMetadata().GetRevision(),
 	}, nil

lib/services/local/scoped_access_test.go

@@ -20,7 +20,7 @@ package local
 
 import (
 	"testing"
-	"time"
+	"testing/synctest"
 
 	"github.com/google/go-cmp/cmp"
 	"github.com/google/uuid"
@@ -42,7 +42,10 @@ import (
 // TestScopedRoleEvents verifies the expected behavior of backend events for the ScopedRole family of types.
 func TestScopedRoleEvents(t *testing.T) {
 	t.Parallel()
+	synctest.Test(t, testScopedRoleEvents)
+}
 
+func testScopedRoleEvents(t *testing.T) {
 	ctx := t.Context()
 
 	backend, err := memory.New(memory.Config{
@@ -70,13 +73,15 @@ func TestScopedRoleEvents(t *testing.T) {
 	defer watcher.Close()
 
 	getNextEvent := func() types.Event {
+		t.Helper()
+		synctest.Wait()
 		select {
 		case event := <-watcher.Events():
 			return event
 		case <-watcher.Done():
 			require.FailNow(t, "Watcher exited with error", watcher.Error())
-		case <-time.After(time.Second * 5):
-			require.FailNow(t, "Timeout waiting for event", watcher.Error())
+		default:
+			require.FailNow(t, "No event ready, synctest bubble is durably blocked")
 		}
 
 		panic("unreachable")
@@ -134,7 +139,8 @@ func TestScopedRoleEvents(t *testing.T) {
 	_ = getNextEvent() // drain the role create event
 
 	assignment := &scopedaccessv1.ScopedRoleAssignment{
-		Kind: scopedaccess.KindScopedRoleAssignment,
+		Kind:    scopedaccess.KindScopedRoleAssignment,
+		SubKind: scopedaccess.SubKindDynamic,
 		Metadata: &headerv1.Metadata{
 			Name: uuid.New().String(),
 		},
@@ -166,7 +172,8 @@ func TestScopedRoleEvents(t *testing.T) {
 
 	// delete the assignment and verify delete event is well-formed.
 	_, err = service.DeleteScopedRoleAssignment(ctx, &scopedaccessv1.DeleteScopedRoleAssignmentRequest{
-		Name: assignment.Metadata.Name,
+		Name:    assignment.Metadata.Name,
+		SubKind: assignment.SubKind,
 	})
 	require.NoError(t, err)
 
@@ -174,11 +181,27 @@ func TestScopedRoleEvents(t *testing.T) {
 	require.Equal(t, types.OpDelete, event.Type)
 
 	require.Empty(t, cmp.Diff(&types.ResourceHeader{
-		Kind: scopedaccess.KindScopedRoleAssignment,
+		Kind:    scopedaccess.KindScopedRoleAssignment,
+		SubKind: scopedaccess.SubKindDynamic,
 		Metadata: types.Metadata{
 			Name: assignment.Metadata.Name,
 		},
 	}, event.Resource.(*types.ResourceHeader), protocmp.Transform()))
+
+	// Assert that any materialized assignments put into the backend (possibly
+	// by an auth service on a later version) don't make it into the event
+	// stream. Use the backend directly to skip subkind validation.
+	assignment.SubKind = scopedaccess.SubKindMaterialized
+	item, err := scopedRoleAssignmentToItem(assignment)
+	require.NoError(t, err)
+	_, err = service.bk.Put(ctx, item)
+	require.NoError(t, err)
+	synctest.Wait()
+	select {
+	case evt := <-watcher.Events():
+		t.Fatalf("expected no event, got %v", evt)
+	default:
+	}
 }
 
 // TestScopedRoleBasicCRUD tests the basic CRUD operations of the ScopedAccessService, excluding the more non-trivial
@@ -427,7 +450,8 @@ func TestScopedRoleAssignmentBasicCRD(t *testing.T) {
 	// basic root assignment to test standard CRD operations with (initially invalid,
 	// will be modified later to be valid)
 	assignment01 := &scopedaccessv1.ScopedRoleAssignment{
-		Kind: scopedaccess.KindScopedRoleAssignment,
+		Kind:    scopedaccess.KindScopedRoleAssignment,
+		SubKind: scopedaccess.SubKindDynamic,
 		Metadata: &headerv1.Metadata{
 			Name: uuid.New().String(),
 		},
@@ -477,7 +501,41 @@ func TestScopedRoleAssignmentBasicCRD(t *testing.T) {
 	require.Error(t, err)
 	require.True(t, trace.IsCompareFailed(err), "expected CompareFailed error, got %v", err)
 
+	// check that otherwise valid assignment fails if subkind is unset.
+	assignment01.SubKind = ""
+	_, err = service.CreateScopedRoleAssignment(ctx, &scopedaccessv1.CreateScopedRoleAssignmentRequest{
+		Assignment: assignment01,
+		RoleRevisions: map[string]string{
+			"role-01": roleRevisions[0],
+		},
+	})
+	require.Error(t, err)
+	require.True(t, trace.IsBadParameter(err), "expected BadParameter error, got %v", err)
+
+	// check that otherwise valid assignment fails if subkind is materialized.
+	assignment01.SubKind = scopedaccess.SubKindMaterialized
+	_, err = service.CreateScopedRoleAssignment(ctx, &scopedaccessv1.CreateScopedRoleAssignmentRequest{
+		Assignment: assignment01,
+		RoleRevisions: map[string]string{
+			"role-01": roleRevisions[0],
+		},
+	})
+	require.Error(t, err)
+	require.True(t, trace.IsBadParameter(err), "expected BadParameter error, got %v", err)
+
+	// check that otherwise valid assignment fails if subkind is unknown.
+	assignment01.SubKind = "unknown"
+	_, err = service.CreateScopedRoleAssignment(ctx, &scopedaccessv1.CreateScopedRoleAssignmentRequest{
+		Assignment: assignment01,
+		RoleRevisions: map[string]string{
+			"role-01": roleRevisions[0],
+		},
+	})
+	require.Error(t, err)
+	require.True(t, trace.IsBadParameter(err), "expected BadParameter error, got %v", err)
+
 	// check that assignment of correct role with correct revision works
+	assignment01.SubKind = scopedaccess.SubKindDynamic
 	crsp, err := service.CreateScopedRoleAssignment(ctx, &scopedaccessv1.CreateScopedRoleAssignmentRequest{
 		Assignment: assignment01,
 		RoleRevisions: map[string]string{
@@ -488,13 +546,22 @@ func TestScopedRoleAssignmentBasicCRD(t *testing.T) {
 	require.NotEmpty(t, crsp.Assignment.Metadata.Revision)
 	require.Empty(t, cmp.Diff(crsp.Assignment, assignment01, protocmp.Transform(), protocmp.IgnoreFields(&headerv1.Metadata{}, "revision")))
 
-	// Check that the assignment can be retrieved.
+	// check that the assignment can be retrieved.
 	grsp, err := service.GetScopedRoleAssignment(ctx, &scopedaccessv1.GetScopedRoleAssignmentRequest{
-		Name: assignment01.Metadata.Name,
+		Name:    assignment01.Metadata.Name,
+		SubKind: scopedaccess.SubKindDynamic,
 	})
 	require.NoError(t, err)
 	require.Empty(t, cmp.Diff(crsp.Assignment, grsp.Assignment, protocmp.Transform() /* deliberately not ignoring revision */))
 
+	// verify that getting a materialized assignment from the backend is an error.
+	_, err = service.GetScopedRoleAssignment(ctx, &scopedaccessv1.GetScopedRoleAssignmentRequest{
+		Name:    assignment01.Metadata.Name,
+		SubKind: scopedaccess.SubKindMaterialized,
+	})
+	require.Error(t, err)
+	require.True(t, trace.IsBadParameter(err), "expected BadParameter error, got %v", err)
+
 	// verify that create fails if the assignment already exists
 	_, err = service.CreateScopedRoleAssignment(ctx, &scopedaccessv1.CreateScopedRoleAssignmentRequest{
 		Assignment: assignment01,
@@ -509,27 +576,49 @@ func TestScopedRoleAssignmentBasicCRD(t *testing.T) {
 	_, err = service.DeleteScopedRoleAssignment(ctx, &scopedaccessv1.DeleteScopedRoleAssignmentRequest{
 		Name:     assignment01.Metadata.Name,
 		Revision: roleRevisions[0],
+		SubKind:  crsp.Assignment.SubKind,
 	})
 	require.Error(t, err)
 	require.True(t, trace.IsCompareFailed(err), "expected CompareFailed error, got %v", err)
 
+	// verify that delete of assignment with materialized subkind fails
+	_, err = service.DeleteScopedRoleAssignment(ctx, &scopedaccessv1.DeleteScopedRoleAssignmentRequest{
+		Name:     assignment01.Metadata.Name,
+		Revision: crsp.Assignment.Metadata.Revision,
+		SubKind:  scopedaccess.SubKindMaterialized,
+	})
+	require.Error(t, err)
+	require.True(t, trace.IsBadParameter(err), "expected BadParameter error, got %v", err)
+
+	// verify that delete of assignment with unknown subkind fails
+	_, err = service.DeleteScopedRoleAssignment(ctx, &scopedaccessv1.DeleteScopedRoleAssignmentRequest{
+		Name:     assignment01.Metadata.Name,
+		Revision: crsp.Assignment.Metadata.Revision,
+		SubKind:  "unknown",
+	})
+	require.Error(t, err)
+	require.True(t, trace.IsBadParameter(err), "expected BadParameter error, got %v", err)
+
 	// verify that delete of assignment with correct revision works
 	_, err = service.DeleteScopedRoleAssignment(ctx, &scopedaccessv1.DeleteScopedRoleAssignmentRequest{
 		Name:     assignment01.Metadata.Name,
 		Revision: crsp.Assignment.Metadata.Revision,
+		SubKind:  crsp.Assignment.SubKind,
 	})
 	require.NoError(t, err)
 
 	// verify that the assignment is gone
 	_, err = service.GetScopedRoleAssignment(ctx, &scopedaccessv1.GetScopedRoleAssignmentRequest{
-		Name: assignment01.Metadata.Name,
+		Name:    assignment01.Metadata.Name,
+		SubKind: assignment01.SubKind,
 	})
 	require.Error(t, err)
 	require.True(t, trace.IsNotFound(err), "expected NotFound error, got %v", err)
 
 	// set up a more non-trivial assignment with multiple sub-assignments
 	assignment02 := &scopedaccessv1.ScopedRoleAssignment{
-		Kind: scopedaccess.KindScopedRoleAssignment,
+		Kind:    scopedaccess.KindScopedRoleAssignment,
+		SubKind: scopedaccess.SubKindDynamic,
 		Metadata: &headerv1.Metadata{
 			Name: uuid.New().String(),
 		},
@@ -603,7 +692,8 @@ func TestScopedRoleAssignmentBasicCRD(t *testing.T) {
 
 	// Check that the assignment can be retrieved
 	grsp, err = service.GetScopedRoleAssignment(ctx, &scopedaccessv1.GetScopedRoleAssignmentRequest{
-		Name: assignment02.Metadata.Name,
+		Name:    assignment02.Metadata.Name,
+		SubKind: assignment02.SubKind,
 	})
 	require.NoError(t, err)
 	require.Empty(t, cmp.Diff(crsp.Assignment, grsp.Assignment, protocmp.Transform() /* deliberately not ignoring revision */))
@@ -612,7 +702,8 @@ func TestScopedRoleAssignmentBasicCRD(t *testing.T) {
 	// bug where original impl would construct invalid conditional actions when multiple sub-assignments
 	// are made for the same role).
 	assignment03 := &scopedaccessv1.ScopedRoleAssignment{
-		Kind: scopedaccess.KindScopedRoleAssignment,
+		Kind:    scopedaccess.KindScopedRoleAssignment,
+		SubKind: scopedaccess.SubKindDynamic,
 		Metadata: &headerv1.Metadata{
 			Name: uuid.New().String(),
 		},
@@ -644,7 +735,8 @@ func TestScopedRoleAssignmentBasicCRD(t *testing.T) {
 
 	// verify that deletion of assignment works
 	_, err = service.DeleteScopedRoleAssignment(ctx, &scopedaccessv1.DeleteScopedRoleAssignmentRequest{
-		Name: assignment03.Metadata.Name,
+		Name:    assignment03.Metadata.Name,
+		SubKind: assignment03.SubKind,
 	})
 	require.NoError(t, err)
 }
@@ -726,7 +818,8 @@ func TestScopedRoleAssignmentInteraction(t *testing.T) {
 
 	// set up a non-trivial assignment with multiple sub-assignments
 	assignment01 := &scopedaccessv1.ScopedRoleAssignment{
-		Kind: scopedaccess.KindScopedRoleAssignment,
+		Kind:    scopedaccess.KindScopedRoleAssignment,
+		SubKind: scopedaccess.SubKindDynamic,
 		Metadata: &headerv1.Metadata{
 			Name: uuid.New().String(),
 		},
@@ -800,6 +893,7 @@ func TestScopedRoleAssignmentInteraction(t *testing.T) {
 	_, err = service.DeleteScopedRoleAssignment(ctx, &scopedaccessv1.DeleteScopedRoleAssignmentRequest{
 		Name:     assignment01.Metadata.Name,
 		Revision: crsp.Assignment.Metadata.Revision,
+		SubKind:  assignment01.SubKind,
 	})
 	require.NoError(t, err)
 

rfd/0243-scoped-roles-in-access-lists.md

@@ -332,7 +332,7 @@ assignments as usual.
 
 We will introduce `sub_kind: materialized` to distinguish materialized
 scoped_role_assignments from user-created ones and prevent name collisions.
-We will use `sub_kind: static` for the current user or automation-created
+We will use `sub_kind: dynamic` for the current user or automation-created
 scoped_role_assignments (this will be a breaking change that has been deemed
 acceptable at this early stage).
 Scoped role assignments will use `<name>/<sub_kind>` as a primary key in the

tool/tctl/common/resource_command_test.go

@@ -340,6 +340,7 @@ spec:
 version: v1
 `
 	const scopedRoleAssignmentYAML = `kind: scoped_role_assignment
+sub_kind: dynamic
 scope: "/"
 spec:
   user: "bob"
@@ -425,10 +426,20 @@ version: v1
 	require.Empty(t, cmp.Diff(expected, rs[0], protocmp.Transform(), protocmp.IgnoreFields(&headerv1.Metadata{}, "revision")))
 
 	// now that a role exists, test commands for assignment creation
-	scopedRoleAssignmentYAMLPath := filepath.Join(t.TempDir(), "some-role-assignment.yaml")
-	require.NoError(t, os.WriteFile(scopedRoleAssignmentYAMLPath, []byte(scopedRoleAssignmentYAML), 0644))
 
-	// Create the scoped role assignment
+	// Can't create an assignment without a subkind.
+	scopedRoleAssignmentYAMLPath := filepath.Join(t.TempDir(), "some-role-assignment.yaml")
+	require.NoError(t, os.WriteFile(
+		scopedRoleAssignmentYAMLPath,
+		[]byte(strings.ReplaceAll(scopedRoleAssignmentYAML, "sub_kind: dynamic\n", "")),
+		0644,
+	))
+	_, err = runResourceCommand(t, clt, []string{"create", scopedRoleAssignmentYAMLPath})
+	require.True(t, trace.IsBadParameter(err), "expected BadParameter error, got %v", err)
+	require.ErrorContains(t, err, "has empty sub_kind")
+
+	// Create the valid scoped role assignment
+	require.NoError(t, os.WriteFile(scopedRoleAssignmentYAMLPath, []byte(scopedRoleAssignmentYAML), 0644))
 	_, err = runResourceCommand(t, clt, []string{"create", scopedRoleAssignmentYAMLPath})
 	require.NoError(t, err)
 
@@ -459,7 +470,11 @@ version: v1
 	require.Len(t, as, 1)
 	assignmentName := as[0].GetMetadata().GetName()
 
-	// Ensure that retrieving the scoped role assignment by name works
+	// Ensure that retrieving the scoped role assignment with incorrect sub_kind fails.
+	_, err = runResourceCommand(t, clt, []string{"get", "scoped_role_assignment/materialized/" + assignmentName, "--format=json"})
+	require.True(t, trace.IsNotFound(err), "expected NotFound error, got %v", err)
+
+	// Ensure that retrieving the scoped role assignment by name with default sub_kind works.
 	buff, err := runResourceCommand(t, clt, []string{"get", "scoped_role_assignment/" + assignmentName, "--format=json"})
 	require.NoError(t, err)
 	var asByName []*scopedaccessv1.ScopedRoleAssignment
@@ -468,9 +483,18 @@ version: v1
 	require.Len(t, asByName, 1)
 	require.Equal(t, assignmentName, asByName[0].GetMetadata().GetName())
 
+	// Ensure that retrieving the scoped role assignment by name with explicit sub_kind works.
+	buff, err = runResourceCommand(t, clt, []string{"get", "scoped_role_assignment/dynamic/" + assignmentName, "--format=json"})
+	require.NoError(t, err)
+	err = json.Unmarshal(buff.Bytes(), &asByName)
+	require.NoError(t, err)
+	require.Len(t, asByName, 1)
+	require.Equal(t, assignmentName, asByName[0].GetMetadata().GetName())
+
 	// Compare with expected value
 	expectedAssignment := &scopedaccessv1.ScopedRoleAssignment{
-		Kind: scopedaccess.KindScopedRoleAssignment,
+		Kind:    scopedaccess.KindScopedRoleAssignment,
+		SubKind: scopedaccess.SubKindDynamic,
 		Metadata: &headerv1.Metadata{
 			Name: assignmentName,
 		},
@@ -490,7 +514,7 @@ version: v1
 	require.Empty(t, cmp.Diff(expectedAssignment, as[0], protocmp.Transform(), protocmp.IgnoreFields(&headerv1.Metadata{}, "revision")))
 
 	// verify delete of assignment
-	_, err = runResourceCommand(t, clt, []string{"rm", "scoped_role_assignment/" + assignmentName})
+	_, err = runResourceCommand(t, clt, []string{"rm", "scoped_role_assignment/dynamic/" + assignmentName})
 	require.NoError(t, err)
 
 	// wait for delete cache propagation

tool/tctl/common/resources/scoped_role_assignment.go

@@ -16,6 +16,7 @@
 package resources
 
 import (
+	"cmp"
 	"context"
 	"fmt"
 	"io"
@@ -52,7 +53,7 @@ func (c *scopedRoleAssignmentCollection) Resources() []types.Resource {
 }
 
 func (c *scopedRoleAssignmentCollection) WriteText(w io.Writer, verbose bool) error {
-	headers := []string{"Scope", "Name", "User", "Assigns"}
+	headers := []string{"SubKind", "Scope", "Name", "User", "Assigns"}
 	rows := make([][]string, len(c.roleAssignments))
 
 	for i, item := range c.roleAssignments {
@@ -62,6 +63,7 @@ func (c *scopedRoleAssignmentCollection) WriteText(w io.Writer, verbose bool) er
 		}
 
 		rows[i] = []string{
+			item.GetSubKind(),
 			item.GetScope(),
 			item.GetMetadata().GetName(),
 			item.GetSpec().GetUser(),
@@ -112,8 +114,11 @@ func createScopedRoleAssignment(ctx context.Context, client *authclient.Client,
 
 func getScopedRoleAssignment(ctx context.Context, client *authclient.Client, ref services.Ref, opts GetOpts) (Collection, error) {
 	if ref.Name != "" {
+		// Default to dynamic if the user didn't specify a subkind.
+		subKind := cmp.Or(ref.SubKind, scopedaccess.SubKindDynamic)
 		rsp, err := client.ScopedAccessServiceClient().GetScopedRoleAssignment(ctx, &scopedaccessv1.GetScopedRoleAssignmentRequest{
-			Name: ref.Name,
+			Name:    ref.Name,
+			SubKind: subKind,
 		})
 		if err != nil {
 			return nil, trace.Wrap(err)
@@ -131,7 +136,8 @@ func getScopedRoleAssignment(ctx context.Context, client *authclient.Client, ref
 
 func deleteScopedRoleAssignment(ctx context.Context, client *authclient.Client, ref services.Ref) error {
 	if _, err := client.ScopedAccessServiceClient().DeleteScopedRoleAssignment(ctx, &scopedaccessv1.DeleteScopedRoleAssignmentRequest{
-		Name: ref.Name,
+		Name:    ref.Name,
+		SubKind: ref.SubKind,
 	}); err != nil {
 		return trace.Wrap(err)
 	}