kvserver: prevent split at invalid tenant prefix keys

Closes #104796.

Epic: None
Release note (bug fix): prevents invalid splits that can crash (and prevent
restarts) of nodes that hold a replica for the right-hand side.

pkg/kv/kvserver/BUILD.bazel

@@ -235,6 +235,7 @@ go_test(
         "batch_spanset_test.go",
         "below_raft_protos_test.go",
         "client_atomic_membership_change_test.go",
+        "client_invalidsplit_test.go",
         "client_lease_test.go",
         "client_merge_test.go",
         "client_metrics_test.go",

pkg/kv/kvserver/client_invalidsplit_test.go

@@ -0,0 +1,49 @@
+// Copyright 2023 The Cockroach Authors.
+//
+// Use of this software is governed by the Business Source License
+// included in the file licenses/BSL.txt.
+//
+// As of the Change Date specified in that file, in accordance with
+// the Business Source License, use of this software will be governed
+// by the Apache License, Version 2.0, included in the file
+// licenses/APL.txt.
+
+package kvserver_test
+
+import (
+	"context"
+	"testing"
+
+	"github.com/cockroachdb/cockroach/pkg/base"
+	"github.com/cockroachdb/cockroach/pkg/keys"
+	"github.com/cockroachdb/cockroach/pkg/testutils/testcluster"
+	"github.com/cockroachdb/cockroach/pkg/util/leaktest"
+	"github.com/cockroachdb/cockroach/pkg/util/log"
+	"github.com/stretchr/testify/require"
+)
+
+func TestSplitAtInvalidTenantPrefix(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+	defer log.Scope(t).Close(t)
+
+	// badKey is the tenant prefix followed by a "large" byte that indicates
+	// that it should be followed by a separate uvarint encoded key (which is
+	// not there).
+	//
+	// See: https://github.com/cockroachdb/cockroach/issues/104796
+	var badKey = append([]byte{'\xfe'}, '\xfd')
+	_, _, err := keys.DecodeTenantPrefix(badKey)
+	t.Log(err)
+	require.Error(t, err)
+
+	ctx := context.Background()
+
+	tc := testcluster.StartTestCluster(t, 1, base.TestClusterArgs{
+		ReplicationMode: base.ReplicationManual,
+	})
+	defer tc.Stopper().Stop(ctx)
+
+	_, _, err = tc.SplitRange(badKey)
+	require.Error(t, err)
+	require.Contains(t, err.Error(), `checking for valid tenantID`)
+}

pkg/kv/kvserver/replica_command.go

@@ -365,6 +365,9 @@ func (r *Replica) adminSplitWithDescriptor(
 		if !storage.IsValidSplitKey(foundSplitKey) {
 			return reply, errors.Errorf("cannot split range at key %s", splitKey)
 		}
+		if _, _, err := keys.DecodeTenantPrefixE(splitKey.AsRawKey()); err != nil {
+			return reply, errors.Wrapf(err, "checking for valid tenantID")
+		}
 	}
 
 	// If the range starts at the splitKey, we treat the AdminSplit