tenantcapabilitiesauthorizer: introduce kill-switch cluster setting

This patch introduces a new (non-public) cluster setting called
`tenant_capabilities.authorizer.enabled` which acts as a kill switch
for the Authorizer. It is inteded to be a safety switch to turn off
capability checks performed by the Authorizer, in case we need to turn
off the subsystem. As such, it isn't inteded to be used during normal
cluster operation.

Release note: None

pkg/multitenant/tenantcapabilities/tenantcapabilitiesauthorizer/BUILD.bazel

@@ -13,6 +13,7 @@ go_library(
         "//pkg/kv/kvpb",
         "//pkg/multitenant/tenantcapabilities",
         "//pkg/roachpb",
+        "//pkg/settings",
         "//pkg/settings/cluster",
         "//pkg/util/log",
         "@com_github_cockroachdb_errors//:errors",
@@ -29,10 +30,12 @@ go_test(
         "//pkg/multitenant/tenantcapabilities",
         "//pkg/multitenant/tenantcapabilities/tenantcapabilitiestestutils",
         "//pkg/roachpb",
+        "//pkg/settings",
         "//pkg/settings/cluster",
         "//pkg/testutils/datapathutils",
         "//pkg/util/leaktest",
         "@com_github_cockroachdb_datadriven//:datadriven",
+        "@com_github_stretchr_testify//require",
     ],
 )
 

pkg/multitenant/tenantcapabilities/tenantcapabilitiesauthorizer/authorizer.go

@@ -16,11 +16,24 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/kv/kvpb"
 	"github.com/cockroachdb/cockroach/pkg/multitenant/tenantcapabilities"
 	"github.com/cockroachdb/cockroach/pkg/roachpb"
+	"github.com/cockroachdb/cockroach/pkg/settings"
 	"github.com/cockroachdb/cockroach/pkg/settings/cluster"
 	"github.com/cockroachdb/cockroach/pkg/util/log"
 	"github.com/cockroachdb/errors"
 )
 
+// authorizerEnabled dictates whether the Authorizer performs any capability
+// checks or not. It is intended as an escape hatch to turn off the tenant
+// capabilities infrastructure; as such, it is intended to be a sort of hammer
+// of last resort, and isn't expected to be used during normal cluster
+// operation.
+var authorizerEnabled = settings.RegisterBoolSetting(
+	settings.SystemOnly,
+	"tenant_capabilities.authorizer.enabled",
+	"enables authorization based on capability checks for incoming (secondary tenant) requests",
+	true,
+)
+
 // Authorizer is a concrete implementation of the tenantcapabilities.Authorizer
 // interface. It's safe for concurrent use.
 type Authorizer struct {
@@ -50,8 +63,8 @@ func New(settings *cluster.Settings, knobs *tenantcapabilities.TestingKnobs) *Au
 func (a *Authorizer) HasCapabilityForBatch(
 	ctx context.Context, tenID roachpb.TenantID, ba *kvpb.BatchRequest,
 ) error {
-	if tenID.IsSystem() {
-		return nil // the system tenant is allowed to do as it pleases.
+	if a.elideCapabilityChecks(ctx, tenID) {
+		return nil
 	}
 	if a.capabilitiesReader == nil {
 		return errors.AssertionFailedf("programming error: trying to perform capability check when no reader exists")
@@ -156,8 +169,8 @@ func (a *Authorizer) BindReader(reader tenantcapabilities.Reader) {
 }
 
 func (a *Authorizer) HasNodeStatusCapability(ctx context.Context, tenID roachpb.TenantID) error {
-	if tenID.IsSystem() {
-		return nil // the system tenant is allowed to do as it pleases.
+	if a.elideCapabilityChecks(ctx, tenID) {
+		return nil
 	}
 	cp, found := a.capabilitiesReader.GetCapabilities(tenID)
 	if !found {
@@ -173,8 +186,8 @@ func (a *Authorizer) HasNodeStatusCapability(ctx context.Context, tenID roachpb.
 }
 
 func (a *Authorizer) HasTSDBQueryCapability(ctx context.Context, tenID roachpb.TenantID) error {
-	if tenID.IsSystem() {
-		return nil // the system tenant is allowed to do as it pleases.
+	if a.elideCapabilityChecks(ctx, tenID) {
+		return nil
 	}
 	cp, found := a.capabilitiesReader.GetCapabilities(tenID)
 	if !found {
@@ -188,3 +201,16 @@ func (a *Authorizer) HasTSDBQueryCapability(ctx context.Context, tenID roachpb.T
 	}
 	return nil
 }
+
+// elideCapabilityChecks returns true if capability checks should be skipped for
+// the supplied tenant.
+func (a *Authorizer) elideCapabilityChecks(ctx context.Context, tenID roachpb.TenantID) bool {
+	if tenID.IsSystem() {
+		return true // the system tenant is allowed to do as it pleases
+	}
+	if !authorizerEnabled.Get(&a.settings.SV) {
+		log.VInfof(ctx, 3, "authorizer turned off; eliding capability checks")
+		return true
+	}
+	return false
+}

pkg/multitenant/tenantcapabilities/tenantcapabilitiesauthorizer/authorizer_test.go

@@ -13,15 +13,18 @@ package tenantcapabilitiesauthorizer
 import (
 	"context"
 	"fmt"
+	"strconv"
 	"testing"
 
 	"github.com/cockroachdb/cockroach/pkg/multitenant/tenantcapabilities"
 	"github.com/cockroachdb/cockroach/pkg/multitenant/tenantcapabilities/tenantcapabilitiestestutils"
 	"github.com/cockroachdb/cockroach/pkg/roachpb"
+	"github.com/cockroachdb/cockroach/pkg/settings"
 	"github.com/cockroachdb/cockroach/pkg/settings/cluster"
 	"github.com/cockroachdb/cockroach/pkg/testutils/datapathutils"
 	"github.com/cockroachdb/cockroach/pkg/util/leaktest"
 	"github.com/cockroachdb/datadriven"
+	"github.com/stretchr/testify/require"
 )
 
 // TestDataDriven runs datadriven tests against the Authorizer interface. The
@@ -52,24 +55,35 @@ import (
 // ----
 // ok
 //
-//
 // "has-tsdb-query-capability": performas a capability check to be able to
 // make TSDB queries. Example:
 //
 // has-tsdb-query-capability ten=11
 // ----
 // ok
-
+//
+// "set-bool-cluster-setting": overrides the specified boolean cluster setting
+// to the given value. Currently, only the authorizerEnabled cluster setting is
+// supported.
+//
+// set-bool-cluster-setting name=tenant_capabilities.authorizer.enabled value=false
+// ----
+// ok
 func TestDataDriven(t *testing.T) {
 	defer leaktest.AfterTest(t)()
 
 	datadriven.Walk(t, datapathutils.TestDataPath(t), func(t *testing.T, path string) {
+		clusterSettings := cluster.MakeTestingClusterSettings()
+		ctx := context.Background()
 		mockReader := mockReader(make(map[roachpb.TenantID]tenantcapabilities.TenantCapabilities))
-		authorizer := New(cluster.MakeTestingClusterSettings(), nil /* TestingKnobs */)
+		authorizer := New(clusterSettings, nil /* TestingKnobs */)
 		authorizer.BindReader(mockReader)
 
 		datadriven.RunTest(t, path, func(t *testing.T, d *datadriven.TestData) string {
-			tenID := tenantcapabilitiestestutils.GetTenantID(t, d)
+			var tenID roachpb.TenantID
+			if d.HasArg("ten") {
+				tenID = tenantcapabilitiestestutils.GetTenantID(t, d)
+			}
 			switch d.Cmd {
 			case "upsert":
 				_, caps, err := tenantcapabilitiestestutils.ParseTenantCapabilityUpsert(t, d)
@@ -106,6 +120,18 @@ func TestDataDriven(t *testing.T) {
 					return "ok"
 				}
 				return err.Error()
+			case "set-bool-cluster-setting":
+				var settingName string
+				d.ScanArgs(t, "name", &settingName)
+				setting, ok := supportedClusterSettings[settingName]
+				if !ok {
+					t.Fatalf("cluster setting %s not supported", settingName)
+				}
+				var valStr string
+				d.ScanArgs(t, "value", &valStr)
+				val, err := strconv.ParseBool(valStr)
+				require.NoError(t, err)
+				setting.Override(ctx, &clusterSettings.SV, val)
 			default:
 				return fmt.Sprintf("unknown command %s", d.Cmd)
 			}
@@ -114,6 +140,13 @@ func TestDataDriven(t *testing.T) {
 	})
 }
 
+// supportedClusterSettings is a map, keyed by cluster setting name, of all
+// boolean cluster settings that can be altered when running datadriven tests
+// for the Authorizer.
+var supportedClusterSettings = map[string]*settings.BoolSetting{
+	authorizerEnabled.Key(): authorizerEnabled,
+}
+
 type mockReader map[roachpb.TenantID]tenantcapabilities.TenantCapabilities
 
 var _ tenantcapabilities.Reader = mockReader{}

pkg/multitenant/tenantcapabilities/tenantcapabilitiesauthorizer/testdata/authorizer_enabled

@@ -0,0 +1,53 @@
+upsert ten=10 can_admin_scatter=false can_admin_split=false can_view_node_info=false can_view_tsdb_metrics=false
+----
+ok
+
+has-capability-for-batch ten=10 cmds=(AdminScatter, Scan)
+----
+client tenant does not have capability "can_admin_scatter" (*kvpb.AdminScatterRequest)
+
+has-node-status-capability ten=10
+----
+client tenant does not have capability to query cluster node metadata
+
+has-tsdb-query-capability ten=10
+----
+client tenant does not have capability to query timeseries data
+
+# Disable the Authorizer.
+set-bool-cluster-setting name=tenant_capabilities.authorizer.enabled value=false
+----
+ok
+
+# Now that the Authorizer is disabled, all the checks that were previously
+# failing should continue to fail.
+
+has-capability-for-batch ten=10 cmds=(AdminScatter, Scan)
+----
+ok
+
+has-node-status-capability ten=10
+----
+ok
+
+has-tsdb-query-capability ten=10
+----
+ok
+
+# Enable the Authorizer again and ensure we start failing capability checks
+# again.
+set-bool-cluster-setting name=tenant_capabilities.authorizer.enabled value=true
+----
+ok
+
+has-capability-for-batch ten=10 cmds=(AdminScatter, Scan)
+----
+client tenant does not have capability "can_admin_scatter" (*kvpb.AdminScatterRequest)
+
+has-node-status-capability ten=10
+----
+client tenant does not have capability to query cluster node metadata
+
+has-tsdb-query-capability ten=10
+----
+client tenant does not have capability to query timeseries data