make scoped key encoding opaque (#68913) (#69544)

* make scoped key encoding opaque

* fix bot naming

build.assets/tooling/go.mod

@@ -293,6 +293,7 @@ require (
 	k8s.io/utils v0.0.0-20251002143259-bc988d571ff4 // indirect
 	mvdan.cc/sh/v3 v3.7.0 // indirect
 	pluginrpc.com/pluginrpc v0.5.0 // indirect
+	rsc.io/ordered v1.1.1 // indirect
 	sigs.k8s.io/controller-runtime v0.20.4 // indirect
 	sigs.k8s.io/json v0.0.0-20250730193827-2d320260d730 // indirect
 	sigs.k8s.io/randfill v1.0.0 // indirect

e2e/runner/go.mod

@@ -287,6 +287,7 @@ require (
 	k8s.io/kube-openapi v0.0.0-20250910181357-589584f1c912 // indirect
 	k8s.io/utils v0.0.0-20251002143259-bc988d571ff4 // indirect
 	mvdan.cc/sh/v3 v3.7.0 // indirect
+	rsc.io/ordered v1.1.1 // indirect
 	sigs.k8s.io/controller-runtime v0.22.4 // indirect
 	sigs.k8s.io/json v0.0.0-20250730193827-2d320260d730 // indirect
 	sigs.k8s.io/randfill v1.0.0 // indirect

lib/auth/machineid/machineidv1/machineidv1_test.go

@@ -671,9 +671,9 @@ func TestCreateBot(t *testing.T) {
 				},
 				Scope: "/scopes/granted",
 				Spec:  &machineidv1pb.BotSpec{},
-				Status: &machineidv1pb.BotStatus{
-					UserName: "bot-++scopes+granted+scoped-bot-success",
-				},
+				Status: machineidv1pb.BotStatus_builder{
+					UserName: "bot-30010173636f7065730000016772616e7465640000-scoped-bot-success",
+				}.Build(),
 			},
 		},
 		{
@@ -697,9 +697,9 @@ func TestCreateBot(t *testing.T) {
 				},
 				Scope: "/scopes/granted",
 				Spec:  &machineidv1pb.BotSpec{},
-				Status: &machineidv1pb.BotStatus{
-					UserName: "bot-++scopes+granted+scoped-bot-from-unscoped",
-				},
+				Status: machineidv1pb.BotStatus_builder{
+					UserName: "bot-30010173636f7065730000016772616e7465640000-scoped-bot-from-unscoped",
+				}.Build(),
 			},
 		},
 		{
@@ -1818,9 +1818,9 @@ func TestUpsertBot(t *testing.T) {
 				},
 				Scope: "/scopes/granted",
 				Spec:  &machineidv1pb.BotSpec{},
-				Status: &machineidv1pb.BotStatus{
-					UserName: "bot-++scopes+granted+scoped-upsert-success",
-				},
+				Status: machineidv1pb.BotStatus_builder{
+					UserName: "bot-30010173636f7065730000016772616e7465640000-scoped-upsert-success",
+				}.Build(),
 			},
 		},
 		{
@@ -1844,9 +1844,9 @@ func TestUpsertBot(t *testing.T) {
 				},
 				Scope: "/scopes/granted",
 				Spec:  &machineidv1pb.BotSpec{},
-				Status: &machineidv1pb.BotStatus{
-					UserName: "bot-++scopes+granted+scoped-upsert-from-unscoped",
-				},
+				Status: machineidv1pb.BotStatus_builder{
+					UserName: "bot-30010173636f7065730000016772616e7465640000-scoped-upsert-from-unscoped",
+				}.Build(),
 			},
 		},
 		{
@@ -2866,9 +2866,9 @@ func TestBotScopeNamespacing(t *testing.T) {
 		wantUserName string
 	}{
 		{scope: "", description: "unscoped", wantUserName: "bot-shared-name"},
-		{scope: "/scopes", description: "parent", wantUserName: "bot-++scopes+shared-name"},
-		{scope: "/scopes/alpha", description: "alpha", wantUserName: "bot-++scopes+alpha+shared-name"},
-		{scope: "/scopes/beta", description: "beta", wantUserName: "bot-++scopes+beta+shared-name"},
+		{scope: "/scopes", description: "parent", wantUserName: "bot-30010173636f7065730000-shared-name"},
+		{scope: "/scopes/alpha", description: "alpha", wantUserName: "bot-30010173636f706573000001616c7068610000-shared-name"},
+		{scope: "/scopes/beta", description: "beta", wantUserName: "bot-30010173636f706573000001626574610000-shared-name"},
 	}
 
 	// Creating each must succeed despite the shared name, and each must land
@@ -2926,7 +2926,7 @@ func TestBotScopeNamespacing(t *testing.T) {
 	}.Build())
 	require.NoError(t, err)
 	require.Equal(t, "alpha updated", upserted.GetMetadata().GetDescription())
-	require.Equal(t, "bot-++scopes+alpha+shared-name", upserted.GetStatus().GetUserName())
+	require.Equal(t, "bot-30010173636f706573000001616c7068610000-shared-name", upserted.GetStatus().GetUserName())
 	upsertEvt, ok := emitter.LastEvent().(*apievents.BotCreate)
 	require.True(t, ok, "expected BotCreate event, got %T", emitter.LastEvent())
 	require.Equal(t, "/scopes/alpha", upsertEvt.ResourceMetadata.Scope)
@@ -2951,7 +2951,7 @@ func TestBotScopeNamespacing(t *testing.T) {
 	require.True(t, ok, "expected BotDelete event, got %T", emitter.LastEvent())
 	require.Equal(t, botName, deleteEvt.ResourceMetadata.Name)
 	require.Equal(t, "/scopes/alpha", deleteEvt.ResourceMetadata.Scope)
-	_, err = srv.Auth().GetUser(ctx, "bot-++scopes+alpha+shared-name", false)
+	_, err = srv.Auth().GetUser(ctx, "bot-30010173636f706573000001616c7068610000-shared-name", false)
 	require.True(t, trace.IsNotFound(err), "expected backing user to be deleted, got: %v", err)
 	_, err = getBot("/scopes/alpha")
 	require.True(t, trace.IsNotFound(err), "expected not found, got: %v", err)
@@ -2980,16 +2980,16 @@ func TestBotScopeNamespacing(t *testing.T) {
 	// scoped bot's encoded user name. Upsert must refuse to write through the
 	// collision in either direction rather than clobber the other bot.
 	squatter, err := botSvc.CreateBot(ctx, machineidv1pb.CreateBotRequest_builder{
-		Bot: newBot("++scopes+alpha+victim", "", "unscoped squatter"),
+		Bot: newBot("30010173636f706573000001616c7068610000-victim", "", "unscoped squatter"),
 	}.Build())
 	require.NoError(t, err)
-	require.Equal(t, "bot-++scopes+alpha+victim", squatter.GetStatus().GetUserName())
+	require.Equal(t, "bot-30010173636f706573000001616c7068610000-victim", squatter.GetStatus().GetUserName())
 	_, err = botSvc.UpsertBot(ctx, machineidv1pb.UpsertBotRequest_builder{
 		Bot: newBot("victim", "/scopes/alpha", "clobber attempt"),
 	}.Build())
 	require.True(t, trace.IsAlreadyExists(err), "expected already exists, got: %v", err)
 	got, err := botSvc.GetBot(ctx, machineidv1pb.GetBotRequest_builder{
-		BotName: "++scopes+alpha+victim",
+		BotName: "30010173636f706573000001616c7068610000-victim",
 	}.Build())
 	require.NoError(t, err)
 	require.Empty(t, cmp.Diff(squatter, got, protocmp.Transform()),
@@ -3000,7 +3000,7 @@ func TestBotScopeNamespacing(t *testing.T) {
 	}.Build())
 	require.NoError(t, err)
 	_, err = botSvc.UpsertBot(ctx, machineidv1pb.UpsertBotRequest_builder{
-		Bot: newBot("++scopes+alpha+victim2", "", "clobber attempt"),
+		Bot: newBot("30010173636f706573000001616c7068610000-victim2", "", "clobber attempt"),
 	}.Build())
 	require.True(t, trace.IsAlreadyExists(err), "expected already exists, got: %v", err)
 	got, err = botSvc.GetBot(ctx, machineidv1pb.GetBotRequest_builder{

lib/join/join_bound_keypair_test.go

@@ -1936,7 +1936,7 @@ func TestJoinBoundKeypair_ScopedToken(t *testing.T) {
 	require.NoError(t, err)
 	certIdentity, err := tlsca.FromSubject(parsedCert.Subject, parsedCert.NotAfter)
 	require.NoError(t, err)
-	require.Equal(t, "bot-++test+test-scoped", certIdentity.Username)
+	require.Equal(t, "bot-300101746573740000-test-scoped", certIdentity.Username)
 
 	// The BotInstance should have the scope set. A bot is identified by
 	// (scope, name), so the read must declare the bot's scope to address the
@@ -1971,7 +1971,7 @@ func TestJoinBoundKeypair_ScopedToken(t *testing.T) {
 				BotInstanceID: firstInstance,
 				TokenName:     scopedToken.GetMetadata().GetName(),
 				Method:        scopedToken.GetSpec().GetJoinMethod(),
-				UserName:      "bot-++test+test-scoped",
+				UserName:      "bot-300101746573740000-test-scoped",
 				BotName:       sqn.Name,
 				Scope:         sqn.Scope,
 			},

lib/scopes/encoding.go

@@ -17,106 +17,115 @@
 package scopes
 
 import (
-	"strings"
+	"encoding/hex"
 
 	"github.com/gravitational/trace"
+	"rsc.io/ordered"
 )
 
+// The scope key encoding produces a single opaque, order-preserving backend key
+// segment for a scope. It is used to namespace scoped resources in the backend,
+// e.g. as the <encoded_scope> component of a key like:
+//
+//	/scoped/<kind>/<encoded_scope>/<name>
+//
+// The encoding is built in two phases. First, the scope is encoded as an
+// [ordered] sequence consisting of a leading discriminant that distinguishes
+// scoped from unscoped values, followed by one string element per scope segment:
+//
+//	unscoped ("")    -> ordered.Encode(unscopedDisc)
+//	root     ("/")   -> ordered.Encode(scopedDisc)
+//	"/a"             -> ordered.Encode(scopedDisc, "a")
+//	"/a/b"           -> ordered.Encode(scopedDisc, "a", "b")
+//
+// This encoding was chosen to satisfy several properties simultaneously:
+//
+//   - Order-preserving: a plain byte-sort of the encoded values reproduces
+//     the same sort order as the [Sort] function.
+//
+//  - Scope prefixing: An encoded scope S is a string prefix of any encoded child scope,
+//    but is *not* a prefix of a scope that is a sibling of S with the same leading segment
+//    characters (e.g. EncodeForKey("/staging") is a prefix of EncodeForKey("/staging/west")
+//    but not of EncodeForKey("/stagingwest")).
+//
+//  - Exact prefixing in backend: Appending the backend separator to the encoded scope allows
+//    backend range queries to retrieve *exactly* the set of keys with that exact scope prefix,
+//    without ambiguity.
+//
+//   - Forward-compatible: The encoding scheme can theoretically handle any future extension to
+//     allowed scope characters.
+
 const (
-	// + sorts before all characters that can appear in a scope (see
-	// segmentRegexp), is allowed in backend keys, and is not the backend key
-	// separator.
-	encodedSeparator       = '+'
-	encodedSeparatorString = string(encodedSeparator)
+	// scopeKeyUnscopedDisc is the leading discriminant for the encoding of an
+	// unscoped value (i.e. EncodeForKey("")). It sorts before
+	// scopeKeyScopedDisc so that unscoped values sort before all scoped values.
+	scopeKeyUnscopedDisc = 0
+
+	// scopeKeyScopedDisc is the leading discriminant for the encoding of any
+	// scoped value (including the root scope "/").
+	scopeKeyScopedDisc = 1
 )
 
 // EncodeForKey encodes a scope so that it will be valid for use in a single
 // backend key segment, preserving sort order. If given an empty string, it
 // will return a non-empty encoding that sorts before all valid encoded
 // scopes.
-//
-// This implementation is a placeholder to unblock development, it is intended
-// to be replaced with a better implementation.
 func EncodeForKey(scope string) (string, error) {
-	// The empty scope is encoded as a single encoded separator.
-	//
-	// All other scopes are encoded with an extra separator at the beginning so
-	// that the root scope sorts after the empty scope.
-	// They also have an extra separator added at the end to support prefix and
-	// exact matches.
 	if scope == "" {
-		return encodedSeparatorString, nil
+		return hex.EncodeToString(ordered.Encode(scopeKeyUnscopedDisc)), nil
 	}
 
 	if err := WeakValidate(scope); err != nil {
 		return "", trace.Wrap(err)
 	}
 
-	// Enforce that the scope does not contain any byte that would sort before
-	// or equal to the encoded separator, which would break sorting.
-	// StrongValidate would prevent this, WeakValidate may not.
-	for _, b := range []byte(scope) {
-		if b <= encodedSeparator {
-			return "", trace.BadParameter("scope contains invalid byte %d which would break sorting", b)
-		}
-	}
-
-	// Enforce that the scope does not contain empty segments, which can also
-	// break sorting of composed keys. For example / and /// could encode to:
-	//
-	//   +++/resource
-	//   +++++/resource
-	//
-	// and the sort order would invert depending on whether the /resource
-	// suffix was present.
-	//
-	// Also enforce that each segment does not start with a byte that would
-	// sort before the backend separator, which could break sorting of composed
-	// keys. For example /a and /a/.b could encode to:
-	//
-	//   ++a+/resource
-	//   ++a+.b+/resource
-	//
-	// and the sort order would be inverted because '.' sorts before '/'
-	// This is only an issue at the first byte of a segment.
+	raw := ordered.Encode(scopeKeyScopedDisc)
 	for segment := range DescendingSegments(scope) {
-		if len(segment) == 0 {
-			return "", trace.BadParameter("scope contains empty segment which would break sorting")
-		}
-		if segment[0] < '/' {
-			return "", trace.BadParameter("scope segment starts with invalid byte %d which would break sorting", segment[0])
-		}
+		raw = ordered.Append(raw, segment)
 	}
 
-	// NormalizeForEquality will trim a trailing separator, which may be
-	// necessary for prefix/exact match queries on encoded scopes.
-	encoded := NormalizeForEquality(scope)
-	encoded = strings.ReplaceAll(encoded, separator, encodedSeparatorString)
-
-	encoded = encodedSeparatorString + encoded + encodedSeparatorString
-	return encoded, nil
+	return hex.EncodeToString(raw), nil
 }
 
-// DecodeFromKey decodes a scope encoded by EncodeForKey.
-func DecodeFromKey(encodedScope string) (string, error) {
-	if encodedScope == encodedSeparatorString {
+// DecodeFromKey decodes a scope encoded by [EncodeForKey].
+func DecodeFromKey(encoded string) (string, error) {
+	raw, err := hex.DecodeString(encoded)
+	if err != nil {
+		return "", trace.BadParameter("invalid encoded scope %q: %v", encoded, err)
+	}
+
+	if len(raw) == 0 {
+		return "", trace.BadParameter("invalid empty encoded scope")
+	}
+
+	var disc int
+	rest, err := ordered.DecodePrefix(raw, &disc)
+	if err != nil {
+		return "", trace.BadParameter("malformed encoded scope %q: %v", encoded, err)
+	}
+
+	switch disc {
+	case scopeKeyUnscopedDisc:
+		if len(rest) != 0 {
+			return "", trace.BadParameter("malformed unscoped encoding %q: unexpected trailing data", encoded)
+		}
 		return "", nil
-	}
+	case scopeKeyScopedDisc:
+		var segments []string
+		for len(rest) > 0 {
+			var segment string
+			if rest, err = ordered.DecodePrefix(rest, &segment); err != nil {
+				return "", trace.BadParameter("malformed encoded scope %q: %v", encoded, err)
+			}
+			segments = append(segments, segment)
+		}
 
-	decoded, ok := strings.CutPrefix(encodedScope, encodedSeparatorString)
-	if !ok {
-		return "", trace.BadParameter("encoded scope did not begin with separator")
+		decoded := Join(segments...)
+		if err := WeakValidate(decoded); err != nil {
+			return "", trace.Wrap(err)
+		}
+		return decoded, nil
+	default:
+		return "", trace.BadParameter("invalid scope discriminant %d in encoded scope %q", disc, encoded)
 	}
-	decoded, ok = strings.CutSuffix(decoded, encodedSeparatorString)
-	if !ok {
-		return "", trace.BadParameter("encoded scope did not end with separator")
-	}
-
-	decoded = strings.ReplaceAll(decoded, encodedSeparatorString, separator)
-
-	if err := WeakValidate(decoded); err != nil {
-		return "", trace.Wrap(err)
-	}
-
-	return decoded, nil
 }

lib/scopes/encoding_test.go

@@ -65,16 +65,47 @@ func TestEncodeDecode(t *testing.T) {
 	}
 }
 
+// TestEncodeForKeyExamples runs through some known examples.
+func TestEncodeForKeyExamples(t *testing.T) {
+	t.Parallel()
+
+	tts := []struct {
+		scope string
+		enc   string
+	}{
+		{scope: "", enc: "3000"},  // unscoped
+		{scope: "/", enc: "3001"}, // root
+		{scope: "/staging", enc: "30010173746167696e670000"},
+		{scope: "/staging/west", enc: "30010173746167696e67000001776573740000"},
+		{scope: "/staging/west/testbed", enc: "30010173746167696e6700000177657374000001746573746265640000"},
+		{scope: "/prod", enc: "30010170726f640000"},
+		{scope: "/prod/west", enc: "30010170726f64000001776573740000"},
+	}
+
+	for _, tt := range tts {
+		t.Run(tt.scope, func(t *testing.T) {
+			encoded, err := EncodeForKey(tt.scope)
+			require.NoError(t, err)
+			require.Equal(t, tt.enc, encoded, "encoding of %q mismatch", tt.scope)
+
+			decoded, err := DecodeFromKey(tt.enc)
+			require.NoError(t, err)
+			require.Equal(t, tt.scope, decoded)
+		})
+	}
+}
+
 // TestEncodeForKeyErrors asserts that EncodeForKeys returns an error in expected cases.
 func TestEncodeForKeyErrors(t *testing.T) {
 	t.Parallel()
 
 	for _, tc := range []string{
+		// empty segments break the ordering/prefix guarantees and are rejected.
 		"//",
 		"/aa//",
-		"/.a",
-		"/aa/.b",
-		"/aa/.b",
+		"/a//b",
+		"///",
+		// non-printable bytes are rejected by weak validation.
 		string([]byte{0, 1, 2, 3}),
 	} {
 		_, err := EncodeForKey(tc)
@@ -82,6 +113,28 @@ func TestEncodeForKeyErrors(t *testing.T) {
 	}
 }
 
+// TestEncodeForKeyWeaklyValid asserts that scopes which are weakly valid but not
+// strongly valid (e.g. segments with a leading dot or uppercase characters)
+// still encode and round-trip. The scope key encoding intentionally relies only
+// on weak validation so that it is forward-compatible with future expansions of
+// the valid scope character set.
+func TestEncodeForKeyWeaklyValid(t *testing.T) {
+	t.Parallel()
+
+	for _, tc := range []string{
+		"/.a",
+		"/aa/.b",
+		"/UPPER",
+		"/a/.hidden/b",
+	} {
+		encoded, err := EncodeForKey(tc)
+		require.NoError(t, err, "expected %q to encode", tc)
+		decoded, err := DecodeFromKey(encoded)
+		require.NoError(t, err)
+		require.Equal(t, NormalizeForEquality(tc), decoded)
+	}
+}
+
 // TestEncodedSort asserts that a basic string sort over encoded scopes
 // produces the same ordering as sorting real scopes with [Sort].
 func TestEncodedSort(t *testing.T) {
@@ -159,20 +212,17 @@ func generateScope() string {
 }
 
 // generateSegment generates a scope segment considered valid by EncodeForKey,
-// meaning it must be weakly valid and start with a byte that sorts after '/'.
+// meaning it must be weakly valid. Weak validation accepts any non-space
+// printable ASCII byte (i.e. [33, 126]) that is not a breaking character, at any
+// position within the segment.
 func generateSegment(segmentLen int) string {
 	const (
-		minStartingByte = '/' + 1
-		minByte         = encodedSeparator + 1
-		maxByte         = 126
+		minByte = 33
+		maxByte = 126
 	)
 	b := make([]byte, segmentLen)
 	for i := range segmentLen {
-		if i == 0 {
-			b[i] = randomValidByteInRange(minStartingByte, maxByte)
-		} else {
-			b[i] = randomValidByteInRange(minByte, maxByte)
-		}
+		b[i] = randomValidByteInRange(minByte, maxByte)
 	}
 	return string(b)
 }

lib/scopes/scopes.go

@@ -189,6 +189,10 @@ func strongValidateFormat(noun, value string, shape *regexp.Regexp) error {
 // about them (e.g. due to significant version drift). Prefer using [StrongValidateSegment] for segments received from
 // external sources (e.g. user input or identity provider).
 func WeakValidateSegment(segment string) error {
+	if segment == "" {
+		return trace.BadParameter("segment is empty")
+	}
+
 	// check for spaces and control characters
 	for _, b := range []byte(segment) {
 		if !isNonSpacePrintableASCII(b) {

lib/scopes/scopes_test.go

@@ -46,7 +46,7 @@ func TestValidateSegment(t *testing.T) {
 			name:     "empty segment",
 			segment:  "",
 			strongOk: false,
-			weakOk:   true,
+			weakOk:   false,
 		},
 		{
 			name:     "segment with leading symbol",
@@ -350,9 +350,9 @@ func TestWeakValidate(t *testing.T) {
 			ok:    true,
 		},
 		{
-			name:  "empty segment passes",
+			name:  "empty segment rejected",
 			scope: "/aa//bb/cc",
-			ok:    true,
+			ok:    false,
 		},
 		{
 			name:  "dangling separator passes",

lib/services/bot_instance.go

@@ -261,9 +261,9 @@ func (o *ListBotInstancesRequestOptions) GetFilterFn() func(*machineidv1.BotInst
 // given bot. An empty Scope refers to an unscoped bot.
 //
 // Bots are namespaced by their scope, so a scoped bot's name encodes the scope
-// as well as the bot name, allowing a name to be reused across scopes. The
-// encoded scope ends in a "+" separator, which cannot appear inside a valid
-// scope, so two different scopes cannot yield the same name. Scoped bots are
+// as well as the bot name (bot-<encoded_scope>-<name>), allowing a name to be reused across
+// scopes. An encoded scope only ever contains lowercase alphanumerics, so the "-" separator
+// keeps the two apart and two different scopes cannot yield the same name. Scoped bots are
 // reconstructed from User labels rather than by parsing this name, so it serves
 // only as an identity key.
 func BotResourceName(bot scopes.QualifiedName) (string, error) {
@@ -273,7 +273,7 @@ func BotResourceName(bot scopes.QualifiedName) (string, error) {
 		if err != nil {
 			return "", trace.Wrap(err, "encoding scope for bot resource name")
 		}
-		name = encodedScope + bot.Name
+		name = encodedScope + "-" + bot.Name
 	}
 	return BotUserPrefix + strings.ReplaceAll(name, " ", "-"), nil
 }

lib/services/bot_instance_test.go

@@ -50,7 +50,7 @@ func TestBotResourceName(t *testing.T) {
 			name:      "scoped",
 			scope:     "/staging",
 			botName:   "name",
-			want:      "bot-++staging+name",
+			want:      "bot-30010173746167696e670000-name",
 			assertErr: require.NoError,
 		},
 		{
@@ -59,7 +59,7 @@ func TestBotResourceName(t *testing.T) {
 			name:      "scoped, child scope",
 			scope:     "/staging/eu",
 			botName:   "name",
-			want:      "bot-++staging+eu+name",
+			want:      "bot-30010173746167696e6700000165750000-name",
 			assertErr: require.NoError,
 		},
 		{

lib/services/local/scoped_access.go

@@ -28,6 +28,7 @@ import (
 
 	"github.com/gravitational/teleport"
 	scopedaccessv1 "github.com/gravitational/teleport/api/gen/proto/go/teleport/scopes/access/v1"
+	scopesv1 "github.com/gravitational/teleport/api/gen/proto/go/teleport/scopes/v1"
 	apiutils "github.com/gravitational/teleport/api/utils"
 	"github.com/gravitational/teleport/lib/backend"
 	"github.com/gravitational/teleport/lib/itertools/stream"
@@ -120,8 +121,14 @@ func (s *ScopedAccessService) ListScopedRoles(ctx context.Context, req *scopedac
 		return nil, trace.NotImplemented("pagination is not implemented for direct backend scoped role reads")
 	}
 
+	// use scopedListRange to narrow the read range where permitted by the scope filter.
+	startKey, endKey, err := scopedListRange(scopedRolePrefix, req.GetScopeFilter())
+	if err != nil {
+		return nil, trace.Wrap(err)
+	}
+
 	var out []*scopedaccessv1.ScopedRole
-	for role, err := range s.StreamScopedRoles(ctx) {
+	for role, err := range s.streamScopedRoles(ctx, startKey, endKey) {
 		if err != nil {
 			return nil, trace.Wrap(err)
 		}
@@ -141,11 +148,17 @@ func (s *ScopedAccessService) ListScopedRoles(ctx context.Context, req *scopedac
 // StreamScopedRoles returns a stream of all scoped roles in the backend. Malformed roles are skipped. Returned roles
 // have had weak validation applied.
 func (s *ScopedAccessService) StreamScopedRoles(ctx context.Context) stream.Stream[*scopedaccessv1.ScopedRole] {
+	startKey := scopedRoleWatchPrefix()
+	return s.streamScopedRoles(ctx, startKey, backend.RangeEnd(startKey))
+}
+
+// streamScopedRoles streams scoped roles from the given backend key range. Malformed roles are skipped. Returned
+// roles have had weak validation applied.
+func (s *ScopedAccessService) streamScopedRoles(ctx context.Context, startKey, endKey backend.Key) stream.Stream[*scopedaccessv1.ScopedRole] {
 	return func(yield func(*scopedaccessv1.ScopedRole, error) bool) {
-		startKey := scopedRoleWatchPrefix()
 		params := backend.ItemsParams{
 			StartKey: startKey,
-			EndKey:   backend.RangeEnd(startKey),
+			EndKey:   endKey,
 		}
 
 		for item, err := range s.bk.Items(ctx, params) {
@@ -362,8 +375,14 @@ func (s *ScopedAccessService) ListScopedRoleAssignments(ctx context.Context, req
 		return nil, trace.NotImplemented("pagination is not implemented for direct backend scoped role assignment reads")
 	}
 
+	// use scopedListRange to narrow the read range where permitted by the scope filter.
+	startKey, endKey, err := scopedListRange(scopedRoleAssignmentPrefix, req.GetScopeFilter())
+	if err != nil {
+		return nil, trace.Wrap(err)
+	}
+
 	var out []*scopedaccessv1.ScopedRoleAssignment
-	for assignment, err := range s.StreamScopedRoleAssignments(ctx) {
+	for assignment, err := range s.streamScopedRoleAssignments(ctx, startKey, endKey) {
 		if err != nil {
 			return nil, trace.Wrap(err)
 		}
@@ -387,11 +406,17 @@ func (s *ScopedAccessService) ListScopedRoleAssignments(ctx context.Context, req
 // StreamScopedRoleAssignments returns a stream of all scoped role assignments in the backend. Malformed assignments are skipped.
 // Returned assignments have had weak validation applied.
 func (s *ScopedAccessService) StreamScopedRoleAssignments(ctx context.Context) stream.Stream[*scopedaccessv1.ScopedRoleAssignment] {
+	startKey := scopedRoleAssignmentWatchPrefix()
+	return s.streamScopedRoleAssignments(ctx, startKey, backend.RangeEnd(startKey))
+}
+
+// streamScopedRoleAssignments streams scoped role assignments from the given backend key range. Malformed assignments
+// are skipped. Returned assignments have had weak validation applied.
+func (s *ScopedAccessService) streamScopedRoleAssignments(ctx context.Context, startKey, endKey backend.Key) stream.Stream[*scopedaccessv1.ScopedRoleAssignment] {
 	return func(yield func(*scopedaccessv1.ScopedRoleAssignment, error) bool) {
-		startKey := scopedRoleAssignmentWatchPrefix()
 		params := backend.ItemsParams{
 			StartKey: startKey,
-			EndKey:   backend.RangeEnd(startKey),
+			EndKey:   endKey,
 		}
 
 		for item, err := range s.bk.Items(ctx, params) {
@@ -642,6 +667,32 @@ func scopedRoleAssignmentWatchPrefix() backend.Key {
 	return backend.ExactKey(scopedPrefix, scopedRoleAssignmentPrefix)
 }
 
+// scopedListRange is a helper for optimistically narrowing the backend key range to be scanned when the
+// scope filter is one that is easily expressible as a backend range query.
+func scopedListRange(kindPrefix string, filter *scopesv1.Filter) (startKey, endKey backend.Key, err error) {
+	switch filter.GetMode() {
+	case scopesv1.Mode_MODE_EXACT:
+		encodedScope, err := scopes.EncodeForKey(filter.GetScope())
+		if err != nil {
+			return backend.Key{}, backend.Key{}, trace.Wrap(err)
+		}
+		// ExactKey appends a trailing separator, restricting the prefix to match only this exact scope segment.
+		start := backend.ExactKey(scopedPrefix, kindPrefix, encodedScope)
+		return start, backend.RangeEnd(start), nil
+	case scopesv1.Mode_MODE_DESCENDANTS:
+		encodedScope, err := scopes.EncodeForKey(filter.GetScope())
+		if err != nil {
+			return backend.Key{}, backend.Key{}, trace.Wrap(err)
+		}
+		// NewKey does not append a trailing separator, so the prefix also matches any descendant scopes.
+		start := backend.NewKey(scopedPrefix, kindPrefix, encodedScope)
+		return start, backend.RangeEnd(start), nil
+	default:
+		start := backend.ExactKey(scopedPrefix, kindPrefix)
+		return start, backend.RangeEnd(start), nil
+	}
+}
+
 // verifyKeyScope checks that a scoped resource's scope field agrees with the scope encoded in its
 // backend key, rejecting the resource if they disagree.
 func verifyKeyScope(key, watchPrefix backend.Key, fieldScope string) error {

lib/services/local/scoped_access_test.go

@@ -20,6 +20,9 @@ package local
 
 import (
 	"context"
+	"iter"
+	"slices"
+	"sync/atomic"
 	"testing"
 	"testing/synctest"
 
@@ -1065,3 +1068,167 @@ func TestScopedAccessScopeConflict(t *testing.T) {
 	}.Build())
 	require.NoError(t, err)
 }
+
+// TestScopedAccessRanging verifies that listing of scoped roles/assignments does optimistic backend range
+// optimizations to reduce unnecessary reads when the scope filter is easily expressible as a backend range
+// query.
+func TestScopedAccessRanging(t *testing.T) {
+	t.Parallel()
+
+	ctx := t.Context()
+
+	bk, err := memory.New(memory.Config{Context: ctx})
+	require.NoError(t, err)
+	defer bk.Close()
+
+	// wrap the backend so we can observe how many items each list operation actually scans, and thereby
+	// assert that the range narrowing genuinely happened.
+	countingBK := &itemCountingBackend{Backend: bk}
+	service := NewScopedAccessService(countingBK)
+
+	scopeList := []string{
+		"/",
+		"/aa",
+		"/aa/bb",
+		"/aa/bb/cc",
+		"/aa/bbb",
+		"/aabb",
+		"/aa-bb",
+		"/bb",
+		"/bb/cc",
+	}
+
+	// assignableFor / effectFor produce valid assignable and scope-of-effect values for a resource at
+	// the given scope. Root cannot be used as an assignable scope or scope of effect, so it gets a
+	// non-root descendant instead.
+	assignableFor := func(scope string) []string {
+		if scope == scopes.Root {
+			return []string{"/subscope"}
+		}
+		return []string{scope}
+	}
+	effectFor := func(scope string) string {
+		if scope == scopes.Root {
+			return "/subscope"
+		}
+		return scope
+	}
+
+	// Create one role and one assignment at each scope. Namespacing allows us to reuse the resource names.
+	for _, scope := range scopeList {
+		_, err := service.CreateScopedRole(ctx, scopedaccessv1.CreateScopedRoleRequest_builder{
+			Role: scopedaccessv1.ScopedRole_builder{
+				Kind:     scopedaccess.KindScopedRole,
+				Metadata: headerv1.Metadata_builder{Name: "listrole"}.Build(),
+				Scope:    scope,
+				Spec:     scopedaccessv1.ScopedRoleSpec_builder{AssignableScopes: assignableFor(scope)}.Build(),
+				Version:  types.V1,
+			}.Build(),
+		}.Build())
+		require.NoError(t, err, "creating role at scope %q", scope)
+
+		_, err = service.CreateScopedRoleAssignment(ctx, scopedaccessv1.CreateScopedRoleAssignmentRequest_builder{
+			Assignment: scopedaccessv1.ScopedRoleAssignment_builder{
+				Kind:     scopedaccess.KindScopedRoleAssignment,
+				SubKind:  scopedaccess.SubKindDynamic,
+				Metadata: headerv1.Metadata_builder{Name: uuid.New().String()}.Build(),
+				Scope:    scope,
+				Spec: scopedaccessv1.ScopedRoleAssignmentSpec_builder{
+					User: "alice",
+					Assignments: []*scopedaccessv1.Assignment{
+						scopedaccessv1.Assignment_builder{Role: "/::listrole", Scope: effectFor(scope)}.Build(),
+					},
+				}.Build(),
+				Version: types.V1,
+			}.Build(),
+		}.Build())
+		require.NoError(t, err, "creating assignment at scope %q", scope)
+	}
+
+	cases := []struct {
+		name  string
+		mode  scopesv1.Mode
+		scope string
+		// narrowed indicates whether the filter is expressible as a backend range query. When true, the
+		// list operation must scan only the matching resources; when false, it falls back to scanning the
+		// entire kind and relies on the in-memory MatchScope filter.
+		narrowed bool
+	}{
+		{name: "exact root", mode: scopesv1.Mode_MODE_EXACT, scope: "/", narrowed: true},
+		{name: "exact mid", mode: scopesv1.Mode_MODE_EXACT, scope: "/aa", narrowed: true},
+		{name: "exact deep", mode: scopesv1.Mode_MODE_EXACT, scope: "/aa/bb", narrowed: true},
+		{name: "descendants root", mode: scopesv1.Mode_MODE_DESCENDANTS, scope: "/", narrowed: true},
+		{name: "descendants mid", mode: scopesv1.Mode_MODE_DESCENDANTS, scope: "/aa", narrowed: true},
+		{name: "descendants deep", mode: scopesv1.Mode_MODE_DESCENDANTS, scope: "/aa/bb", narrowed: true},
+		{name: "descendants other subtree", mode: scopesv1.Mode_MODE_DESCENDANTS, scope: "/bb", narrowed: true},
+		{name: "ancestors falls back to full scan", mode: scopesv1.Mode_MODE_ANCESTORS, scope: "/aa/bb", narrowed: false},
+	}
+
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			ctx := t.Context()
+			filter := scopesv1.Filter_builder{Scope: tc.scope, Mode: tc.mode}.Build()
+			require.NoError(t, scopes.ValidateFilter(filter))
+
+			// collect the authoritative set of scopes the filter selects.
+			var matchWant []string
+			for _, scope := range scopeList {
+				if scopes.MatchScope(filter, scope) {
+					matchWant = append(matchWant, scope)
+				}
+			}
+			slices.Sort(matchWant)
+			require.NotEmpty(t, matchWant, "test case should select at least one scope")
+
+			// A narrowed range scans exactly the matching resources; a full scan visits every resource of
+			// the kind. There is exactly one role and one assignment per scope.
+			wantScanned := len(matchWant)
+			if !tc.narrowed {
+				wantScanned = len(scopeList)
+			}
+
+			countingBK.itemsScanned.Store(0)
+			rolesResp, err := service.ListScopedRoles(ctx, scopedaccessv1.ListScopedRolesRequest_builder{ScopeFilter: filter}.Build())
+			require.NoError(t, err)
+			var gotRoleScopes []string
+			for _, role := range rolesResp.GetRoles() {
+				gotRoleScopes = append(gotRoleScopes, role.GetScope())
+			}
+			slices.Sort(gotRoleScopes)
+			require.Equal(t, matchWant, gotRoleScopes, "ListScopedRoles")
+			require.Equal(t, wantScanned, int(countingBK.itemsScanned.Load()), "ListScopedRoles scanned item count")
+
+			countingBK.itemsScanned.Store(0)
+			assignmentsResp, err := service.ListScopedRoleAssignments(ctx, scopedaccessv1.ListScopedRoleAssignmentsRequest_builder{ScopeFilter: filter}.Build())
+			require.NoError(t, err)
+			var gotAssignmentScopes []string
+			for _, assignment := range assignmentsResp.GetAssignments() {
+				gotAssignmentScopes = append(gotAssignmentScopes, assignment.GetScope())
+			}
+			slices.Sort(gotAssignmentScopes)
+			require.Equal(t, matchWant, gotAssignmentScopes, "ListScopedRoleAssignments")
+			require.Equal(t, wantScanned, int(countingBK.itemsScanned.Load()), "ListScopedRoleAssignments scanned item count")
+		})
+	}
+}
+
+// itemCountingBackend wraps a backend.Backend and counts the number of items yielded by Items calls,
+// letting tests observe how many items a range scan actually visited.
+type itemCountingBackend struct {
+	backend.Backend
+	itemsScanned atomic.Int64
+}
+
+func (b *itemCountingBackend) Items(ctx context.Context, params backend.ItemsParams) iter.Seq2[backend.Item, error] {
+	items := b.Backend.Items(ctx, params)
+	return func(yield func(backend.Item, error) bool) {
+		for item, err := range items {
+			if err == nil {
+				b.itemsScanned.Add(1)
+			}
+			if !yield(item, err) {
+				return
+			}
+		}
+	}
+}