kv: reject admin KV requests from tenant SQL processes

Fixes cockroachdb#52360.
First two commits from cockroachdb#52589.

This commit adds a new restriction to the tenantAuth policy that it
accepts no "Admin" KV requests. This prevents tenants from splitting
ranges, merging range, rebalancing ranges, or issuing any other KV
requests with the `isAdmin` flag that could dictate KV-level
distribution decisions.

This further mitigates the impact that a compromised tenant SQL process
could have on the rest of the cluster.

pkg/ccl/kvccl/kvtenantccl/proxy.go

@@ -32,6 +32,8 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/util/syncutil/singleflight"
 	"github.com/cockroachdb/errors"
 	"google.golang.org/grpc"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
 )
 
 func init() {
@@ -315,7 +317,7 @@ func (p *Proxy) RangeLookup(
 
 // FirstRange implements the kvcoord.RangeDescriptorDB interface.
 func (p *Proxy) FirstRange() (*roachpb.RangeDescriptor, error) {
-	return nil, errors.New("kvtenant.Proxy does not have access to FirstRange")
+	return nil, status.Error(codes.Unauthenticated, "kvtenant.Proxy does not have access to FirstRange")
 }
 
 // getClient returns the singleton InternalClient if one is currently active. If

pkg/ccl/kvccl/kvtenantccl/proxy_test.go

@@ -21,6 +21,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/rpc"
 	"github.com/cockroachdb/cockroach/pkg/testutils"
 	"github.com/cockroachdb/cockroach/pkg/util"
+	"github.com/cockroachdb/cockroach/pkg/util/grpcutil"
 	"github.com/cockroachdb/cockroach/pkg/util/hlc"
 	"github.com/cockroachdb/cockroach/pkg/util/leaktest"
 	"github.com/cockroachdb/cockroach/pkg/util/log"
@@ -295,6 +296,7 @@ func TestProxyRangeLookup(t *testing.T) {
 	desc, err := p.FirstRange()
 	require.Nil(t, desc)
 	require.Regexp(t, "does not have access to FirstRange", err)
+	require.True(t, grpcutil.IsAuthenticationError(err))
 }
 
 // TestProxyRetriesUnreachable tests that Proxy iterates over each of its

pkg/ccl/logictestccl/testdata/logic_test/tenant_unsupported

@@ -78,3 +78,29 @@ SHOW ZONE CONFIGURATION FOR TABLE kv
 
 statement error operation is unsupported
 ALTER TABLE kv CONFIGURE ZONE USING num_replicas = 123
+
+# Cannot perform operations that issue Admin requests.
+#
+# TODO DURING REVIEW: we could wrap these up in nicer errors. It's unclear
+# whether doing so is worth the effort or where we would do so if we wanted to.
+
+statement error Unauthenticated desc = admin request \[1 AdmSplit\] not permitted
+ALTER TABLE kv SPLIT AT VALUES ('foo')
+
+statement error Unauthenticated desc = admin request \[1 AdmUnsplit\] not permitted
+ALTER TABLE kv UNSPLIT AT VALUES ('foo')
+
+statement error Unauthenticated desc = requested key /Meta2/"\\x00" not fully contained in tenant keyspace /Tenant/1\{0-1\}
+ALTER TABLE kv UNSPLIT ALL
+
+statement error Unauthenticated desc = admin request \[1 AdmScatter\] not permitted
+ALTER TABLE kv SCATTER
+
+statement error operation is unsupported in multi-tenancy mode
+ALTER TABLE kv EXPERIMENTAL_RELOCATE VALUES (ARRAY[1], 'k')
+
+statement error Unauthenticated desc = requested key /Meta2/Tenant/10/Table/53/1/"k"/0 not fully contained in tenant keyspace /Tenant/1\{0-1\}
+ALTER TABLE kv EXPERIMENTAL_RELOCATE LEASE VALUES (1, 'k')
+
+statement error Unauthenticated desc = kvtenant.Proxy does not have access to FirstRange
+SELECT crdb_internal.check_consistency(true, '', '')

pkg/rpc/auth_tenant.go

@@ -116,6 +116,12 @@ func (a tenantAuth) authRequest(
 // authBatch authorizes the provided tenant to invoke the Batch RPC with the
 // provided args.
 func (a tenantAuth) authBatch(tenID roachpb.TenantID, args *roachpb.BatchRequest) error {
+	// Admin batch requests are not permitted.
+	if args.IsAdmin() {
+		return authErrorf("admin request [%s] not permitted", args.Summary())
+	}
+
+	// All keys in the request must reside within the tenant's keyspace.
 	rSpan, err := keys.Range(args.Requests)
 	if err != nil {
 		return authError(err.Error())

pkg/rpc/auth_test.go

@@ -153,6 +153,11 @@ func TestTenantAuthRequest(t *testing.T) {
 		h := roachpb.RequestHeaderFromSpan(s)
 		return &roachpb.ScanRequest{RequestHeader: h}
 	}
+	makeAdminReq := func(key string) roachpb.Request {
+		s := makeSpan(key)
+		h := roachpb.RequestHeaderFromSpan(s)
+		return &roachpb.AdminSplitRequest{RequestHeader: h, SplitKey: s.Key}
+	}
 	makeReqs := func(reqs ...roachpb.Request) []roachpb.RequestUnion {
 		ru := make([]roachpb.RequestUnion, len(reqs))
 		for i, r := range reqs {
@@ -221,6 +226,38 @@ func TestTenantAuthRequest(t *testing.T) {
 				)},
 				expErr: `requested key span /Tenant/{10"a"-20"b"} not fully contained in tenant keyspace /Tenant/1{0-1}`,
 			},
+			{
+				req: &roachpb.BatchRequest{Requests: makeReqs(
+					makeAdminReq("a"),
+				)},
+				expErr: `admin request \[1 AdmSplit\] not permitted`,
+			},
+			{
+				req: &roachpb.BatchRequest{Requests: makeReqs(
+					makeAdminReq(prefix(10, "a")),
+				)},
+				expErr: `admin request \[1 AdmSplit\] not permitted`,
+			},
+			{
+				req: &roachpb.BatchRequest{Requests: makeReqs(
+					makeAdminReq(prefix(50, "a")),
+				)},
+				expErr: `admin request \[1 AdmSplit\] not permitted`,
+			},
+			{
+				req: &roachpb.BatchRequest{Requests: makeReqs(
+					makeAdminReq(prefix(10, "a")),
+					makeReq(prefix(10, "a"), prefix(10, "b")),
+				)},
+				expErr: `admin request \[1 Scan, 1 AdmSplit\] not permitted`,
+			},
+			{
+				req: &roachpb.BatchRequest{Requests: makeReqs(
+					makeReq(prefix(10, "a"), prefix(10, "b")),
+					makeAdminReq(prefix(10, "a")),
+				)},
+				expErr: `admin request \[1 Scan, 1 AdmSplit\] not permitted`,
+			},
 		},
 		"/cockroach.roachpb.Internal/RangeLookup": {
 			{

pkg/sql/backfill.go

@@ -1294,11 +1294,15 @@ func (sc *SchemaChanger) backfillIndexes(
 		fn()
 	}
 
-	expirationTime := sc.db.Clock().Now().Add(time.Hour.Nanoseconds(), 0)
-
-	for _, span := range addingSpans {
-		if err := sc.db.AdminSplit(ctx, span.Key, expirationTime); err != nil {
-			return err
+	// Split off a new range for each new index span. But only do so for the
+	// system tenant. Secondary tenants do not have mandatory split points
+	// between tables or indexes.
+	if sc.execCfg.Codec.ForSystemTenant() {
+		expirationTime := sc.db.Clock().Now().Add(time.Hour.Nanoseconds(), 0)
+		for _, span := range addingSpans {
+			if err := sc.db.AdminSplit(ctx, span.Key, expirationTime); err != nil {
+				return err
+			}
 		}
 	}
 

pkg/sql/logictest/testdata/logic_test/select_index_span_ranges

@@ -1,3 +1,5 @@
+# LogicTest: !3node-tenant
+
 # This test verifies that we correctly perform an index join when the KV
 # batches span ranges. This is testing that SQL disables batch limits for index
 # join; otherwise it can get out of order results from KV that it can't handle.

pkg/sql/logictest/testdata/logic_test/subquery

@@ -155,8 +155,12 @@ CREATE TABLE abc (a INT PRIMARY KEY, b INT, c INT)
 statement ok
 INSERT INTO abc VALUES (1, 2, 3), (4, 5, 6)
 
-statement ok
-ALTER TABLE abc SPLIT AT VALUES ((SELECT 1))
+# TODO(nvanbenschoten): until we have conditional logic in these files, disable
+# this statement so that we can continue to test this file with the 3node-tenant
+# config.
+#
+# statement ok
+# ALTER TABLE abc SPLIT AT VALUES ((SELECT 1))
 
 query error unsupported comparison operator: <tuple{int, int}> IN <tuple{tuple{int AS a, int AS b, int AS c}}>
 SELECT (1, 2) IN (SELECT * FROM abc)