sql: prevent delegate behavior from being blocked by the unsafesql gate

The epic below outlines a set of work to prevent users from unauthorized
access to what we deem unsafe internals. These unsafe internals lie
mostly in the crdb_internal schema and the system database. This PR
prevents that gate from being checked within delegate query behavior,
thus allowing users to issue delegates from commands like `SHOW
DATABASES`.

Fixes: #151406
Epic: CRDB-24527

Release note (ops change): Delegate queries (like SHOW etc) are excluded
from unsafesql checks to accessing the system or crdb_internal tables. 

Author: angles-n-daemons

pkg/sql/authorization.go

@@ -952,6 +952,10 @@ func (p *planner) HasViewActivityOrViewActivityRedactedRole(
 // objectIsUnsafe checks if the privilege object is considered unsafe for external usage.
 // Unsafe objects are any system tables, and crdb_internal tables which are not listed as externally supported.
 func (p *planner) objectIsUnsafe(ctx context.Context, privilegeObject privilege.Object) bool {
+	if p.skipUnsafeInternalsCheck {
+		return false
+	}
+
 	d, ok := privilegeObject.(catalog.TableDescriptor)
 	if !ok {
 		return true

pkg/sql/authorization_test.go

@@ -7,6 +7,7 @@ package sql_test
 
 import (
 	"context"
+	gosql "database/sql"
 	"fmt"
 	"strings"
 	"sync"
@@ -321,6 +322,14 @@ func TestCheckUnsafeInternalsAccess(t *testing.T) {
 	})
 	defer s.Stopper().Stop(ctx)
 
+	// helper func for setting a safe connection.
+	safeConn := func(t *testing.T) *gosql.DB {
+		conn := s.SQLConn(t)
+		_, err := conn.Exec("SET allow_unsafe_internals = false")
+		require.NoError(t, err)
+		return conn
+	}
+
 	t.Run("accessing the system database", func(t *testing.T) {
 		q := "SELECT * FROM system.namespace"
 
@@ -378,47 +387,66 @@ func TestCheckUnsafeInternalsAccess(t *testing.T) {
 
 	t.Run("accessing the crdb_internal schema", func(t *testing.T) {
 		t.Run("supported table allowed", func(t *testing.T) {
-			conn := s.SQLConn(t)
-			_, err := conn.Exec("SET allow_unsafe_internals = false")
-			require.NoError(t, err)
+			conn := safeConn(t)
 
 			// Supported crdb_internal tables should be allowed even when allow_unsafe_internals = false
-			_, err = conn.Query("SELECT * FROM crdb_internal.zones")
+			_, err := conn.Query("SELECT * FROM crdb_internal.zones")
 			require.NoError(t, err, "supported crdb_internal table (zones) should be accessible when allow_unsafe_internals = false")
 		})
 
 		t.Run("unsupported table denied", func(t *testing.T) {
-			conn := s.SQLConn(t)
-			_, err := conn.Exec("SET allow_unsafe_internals = false")
-			require.NoError(t, err)
+			conn := safeConn(t)
 
 			// Unsupported crdb_internal tables should be denied when allow_unsafe_internals = false
-			_, err = conn.Query("SELECT * FROM crdb_internal.gossip_alerts")
+			_, err := conn.Query("SELECT * FROM crdb_internal.gossip_alerts")
 			checkUnsafeErr(t, err)
 		})
 	})
 
-	// The functionality for this lies in the pkg/sql/optbuilder/scalar.go
-	// file, but it is tested here as that package does not setup a test
-	// server.
+	// The functionality for this lies in the optbuilder package file,
+	// but it is tested here as that package does not setup a test server.
 	t.Run("accessing crdb_internal builtins", func(t *testing.T) {
 		t.Run("non crdb_internal builtin allowed", func(t *testing.T) {
-			conn := s.SQLConn(t)
-			_, err := conn.Exec("SET allow_unsafe_internals = false")
-			require.NoError(t, err)
+			conn := safeConn(t)
 
 			// Non crdb_internal tables should be allowed.
-			_, err = conn.Query("SELECT * FROM generate_series(1,5)")
+			_, err := conn.Query("SELECT * FROM generate_series(1,5)")
 			require.NoError(t, err)
 		})
 
 		t.Run("crdb_internal builtin not allowed", func(t *testing.T) {
-			conn := s.SQLConn(t)
-			_, err := conn.Exec("SET allow_unsafe_internals = false")
-			require.NoError(t, err)
+			conn := safeConn(t)
 
 			// Unsupported crdb_internal builtins should be denied.
-			_, err = conn.Query("SELECT * FROM crdb_internal.tenant_span_stats()")
+			_, err := conn.Query("SELECT * FROM crdb_internal.tenant_span_stats()")
+			checkUnsafeErr(t, err)
+		})
+	})
+
+	// The functionality for this check also lives in the optbuilder package
+	// but is tested here.
+	t.Run("skips delegation", func(t *testing.T) {
+		t.Run("delegation is allowed", func(t *testing.T) {
+			conn := safeConn(t)
+
+			// tests delegation to builtins
+			_, err := conn.Exec("show grants")
+			require.NoError(t, err)
+
+			// tests delegation to crdb_internal tables
+			_, err = conn.Exec("show databases")
+			require.NoError(t, err)
+		})
+
+		t.Run("underlying tables which delegates rely on are not", func(t *testing.T) {
+			conn := safeConn(t)
+
+			// tests delegation to builtins
+			_, err := conn.Exec("SELECT * FROM crdb_internal.privilege_name('DELETE')")
+			checkUnsafeErr(t, err)
+
+			// tests delegation to crdb_internal tables
+			_, err = conn.Exec("SELECT * FROM crdb_internal.databases")
 			checkUnsafeErr(t, err)
 		})
 	})

pkg/sql/opt/cat/catalog.go

@@ -286,4 +286,7 @@ type Catalog interface {
 
 	// IsOwner returns true if user is the owner of the object o
 	IsOwner(ctx context.Context, o Object, user username.SQLUsername) (bool, error)
+
+	// DisableUnsafeInternalsCheck disables the check for unsafe internals in the catalog.
+	DisableUnsafeInternalCheck() func()
 }

pkg/sql/opt/optbuilder/builder.go

@@ -202,6 +202,10 @@ type Builder struct {
 	// built, and can be safely reused across different call-sites within the same
 	// memo.
 	builtTriggerFuncs map[cat.StableID][]cachedTriggerFunc
+
+	// skipUnsafeInternalsCheck is used to skip the check that the
+	// planner is not used for unsafe internal statements.
+	skipUnsafeInternalsCheck bool
 }
 
 // New creates a new Builder structure initialized with the given
@@ -499,6 +503,8 @@ func (b *Builder) buildStmt(
 			// register all those dependencies with the metadata (for cache
 			// invalidation). We don't care about caching plans for these statements.
 			b.DisableMemoReuse = true
+			// It's considered acceptable when we delegate to unsafe internals.
+			defer b.disableUnsafeInternalCheck()()
 			return b.buildStmt(newStmt, desiredTypes, inScope)
 		}
 
@@ -584,6 +590,18 @@ func (b *Builder) maybeTrackUserDefinedTypeDepsForViews(texpr tree.TypedExpr) {
 	}
 }
 
+// disableUnsafeInternalCheck is used to disable the check that the
+// prevents external users from accessing unsafe internals.
+func (b *Builder) disableUnsafeInternalCheck() func() {
+	b.skipUnsafeInternalsCheck = true
+	cleanup := b.catalog.DisableUnsafeInternalCheck()
+
+	return func() {
+		b.skipUnsafeInternalsCheck = false
+		cleanup()
+	}
+}
+
 // optTrackingTypeResolver is a wrapper around a TypeReferenceResolver that
 // remembers all of the resolved types in the provided Metadata.
 type optTrackingTypeResolver struct {

pkg/sql/opt/optbuilder/scalar.go

@@ -890,6 +890,10 @@ func (b *Builder) constructUnary(
 func (b *Builder) isUnsafeBuiltin(
 	overload *tree.Overload, def *tree.ResolvedFunctionDefinition,
 ) bool {
+	if b.skipUnsafeInternalsCheck {
+		return false
+
+	}
 	if overload.Type != tree.BuiltinRoutine {
 		return false
 	}

pkg/sql/opt/testutils/testcat/test_catalog.go

@@ -551,6 +551,11 @@ func (tc *Catalog) ExecuteMultipleDDL(sql string) error {
 	return nil
 }
 
+// DisableUnsafeInternalCheck is part of the cat.Catalog interface.
+func (tc *Catalog) DisableUnsafeInternalCheck() func() {
+	return func() {}
+}
+
 // ExecuteDDL parses the given DDL SQL statement and creates objects in the test
 // catalog. This is used to test without spinning up a cluster.
 func (tc *Catalog) ExecuteDDL(sql string) (string, error) {

pkg/sql/opt_catalog.go

@@ -742,6 +742,15 @@ func (oc *optCatalog) codec() keys.SQLCodec {
 	return oc.planner.ExecCfg().Codec
 }
 
+// DisableUnsafeInternalCheck sets the planners skipUnsafeInternalsCheck
+// to true, and returns a function which reverses it to false.
+func (oc *optCatalog) DisableUnsafeInternalCheck() func() {
+	oc.planner.skipUnsafeInternalsCheck = true
+	return func() {
+		oc.planner.skipUnsafeInternalsCheck = false
+	}
+}
+
 // optView is a wrapper around catalog.TableDescriptor that implements
 // the cat.Object, cat.DataSource, and cat.View interfaces.
 type optView struct {

pkg/sql/planner.go

@@ -305,6 +305,10 @@ type planner struct {
 	// statement. It's similar to autoRetryCounter / txnState.mu.autoRetryCounter
 	// but for statement retries.
 	autoRetryStmtCounter int
+
+	// skipUnsafeInternalsCheck is used to skip the check that the
+	// planner is not used for unsafe internal statements.
+	skipUnsafeInternalsCheck bool
 }
 
 // hasFlowForPausablePortal returns true if the planner is for re-executing a