logictest: flush out indirect unsafe access in logictests

We recently added a session variable which must be explicitly set
for a caller to access system and crdb_internal tables. There's a
concern however that the system will block due to "indirect" access, be
it query expansion, rewriting or anything else.

So users aren't surprised by this behavior, this PR modifies the
logictests so that by default, unless a query explicitly references the
system or crdb_internal tables, it will disallow access to the unsafe
internals. Doing so will hopefully flush out any possible instances of
unsafe access.

This PR also does a few things to make this possible:
 - Introduces a way in tests to override the session variable.
 - Adds a new knob option for the logictests, so that if there's
	 indirect access because a UDF is defined, the test writer can
	 explicitly allow for it.

Fixes: #154675
Fixes: #155367
Epic: CRDB-55276
Release note: None

Author: kyle-a-wong

pkg/ccl/logictestccl/testdata/logic_test/multi_region_survival_goal

@@ -1,4 +1,4 @@
-# knob-opt: sync-event-log
+# knob-opt: sync-event-log allow-unsafe
 # LogicTest: multiregion-9node-3region-3azs-tenant
 
 # Only the root user can modify the system database's regions.

pkg/ccl/logictestccl/testdata/logic_test/plpgsql_txn

@@ -1,4 +1,5 @@
 # LogicTest: !local-prepared
+# knob-opt: allow-unsafe
 
 statement ok
 CREATE TABLE t (x INT);

pkg/ccl/logictestccl/testdata/logic_test/redact_descriptor

@@ -1,4 +1,5 @@
 # LogicTest: !schema-locked-disabled
+# knob-opt: allow-unsafe
 
 # NB: "Xw==" is the byte representation of "_", which is what we use to redact fields.
 statement ok

pkg/ccl/logictestccl/testdata/logic_test/triggers

@@ -1,4 +1,5 @@
 # LogicTest: !local-legacy-schema-changer !local-prepared
+# knob-opt: allow-unsafe
 
 # Include some hidden columns that won't be visible to triggers.
 statement ok

pkg/ccl/logictestccl/testdata/logic_test/udf_rewrite

@@ -1,3 +1,5 @@
+# knob-opt: allow-unsafe
+
 statement ok
 CREATE SEQUENCE seq;
 

pkg/sql/authorization.go

@@ -125,7 +125,8 @@ func (p *planner) HasPrivilege(
 	// Do a safety check on the object, if it is considered unsafe
 	// does the caller have the appropriate session data to access it?
 	if p.objectIsUnsafe(ctx, privilegeObject) {
-		if err := unsafesql.CheckInternalsAccess(ctx, p.SessionData(), p.stmt.AST, p.extendedEvalCtx.Annotations, &p.ExecCfg().Settings.SV); err != nil {
+		unsafeOverride := p.ExecCfg().EvalContextTestingKnobs.UnsafeOverride
+		if err := unsafesql.CheckInternalsAccess(ctx, p.SessionData(), p.stmt.AST, p.extendedEvalCtx.Annotations, &p.ExecCfg().Settings.SV, unsafeOverride); err != nil {
 			return false, err
 		}
 	}
@@ -961,7 +962,7 @@ func (p *planner) objectIsUnsafe(ctx context.Context, privilegeObject privilege.
 		return false
 	}
 
-	// All system descriptors are considered unsafe.
+	// All system descriptors are considered unsafe
 	if catalog.IsSystemDescriptor(d) {
 		return true
 	}

pkg/sql/crdb_internal.go

@@ -4491,6 +4491,13 @@ CREATE TABLE crdb_internal.ranges_no_leases (
 `,
 	resultColumns: colinfo.RangesNoLeases,
 	generator: func(ctx context.Context, p *planner, _ catalog.DatabaseDescriptor, limit int64, _ *stop.Stopper) (virtualTableGenerator, cleanupFunc, error) {
+		// We have to disable the unsafe check on execution, because for this table
+		// we may have gotten to it through delegate execution, eg. SHOW RANGES.
+		// Accessing it will still fail if queried directly during planning, eg
+		// `SELECT * FROM crdb_internal.ranges_no_leases`, so it's safe to unblock
+		// here.
+		defer p.DisableUnsafeInternalsCheck()()
+
 		hasAdmin, err := p.HasAdminRole(ctx)
 		if err != nil {
 			return nil, nil, err

pkg/sql/logictest/logic.go

@@ -330,6 +330,7 @@ import (
 //      - noticetrace: runs the query and compares only the notices that
 //						appear. Cannot be combined with kvtrace.
 //      - nodeidx=N: runs the query on node N of the cluster.
+//      - allowunsafe: allows access to unsafe internals during execution.
 //
 //    The label is optional. If specified, the test runner stores a hash
 //    of the results of the query under the given label. If the label is
@@ -969,6 +970,9 @@ type logicQuery struct {
 	// roundFloatsInStringsSigFigs specifies the number of significant figures
 	// to round floats embedded in strings to where zero means do not round.
 	roundFloatsInStringsSigFigs int
+
+	// allowUnsafe indicates whether unsafe operations are allowed during execution.
+	allowUnsafe bool
 }
 
 var allowedKVOpTypes = []string{
@@ -1119,6 +1123,10 @@ type logicTest struct {
 	// retryDuration is the maximum duration to retry a statement when using
 	// the retry directive.
 	retryDuration time.Duration
+
+	// allowUnsafe is a variable which controls whether the test can access
+	// unsafe internals.
+	allowUnsafe bool
 }
 
 func (t *logicTest) t() *testing.T {
@@ -1493,6 +1501,7 @@ func (t *logicTest) newCluster(
 			DisableOptimizerRuleProbability: *disableOptRuleProbability,
 			OptimizerCostPerturbation:       *optimizerCostPerturbation,
 			ForceProductionValues:           serverArgs.ForceProductionValues,
+			UnsafeOverride:                  func() *bool { return &t.allowUnsafe },
 		}
 		knobs.SQLExecutor = &sql.ExecutorTestingKnobs{
 			DeterministicExplain:            true,
@@ -2115,6 +2124,18 @@ func (c knobOptSynchronousEventLog) apply(args *base.TestingKnobs) {
 	args.EventLog.(*eventlog.EventLogTestingKnobs).SyncWrites = true
 }
 
+// knobOptAllowUnsafe always allows access to the unsafe internals.
+type knobOptAllowUnsafe struct{}
+
+var _ knobOpt = knobOptAllowUnsafe{}
+
+// apply implements the clusterOpt interface.
+func (c knobOptAllowUnsafe) apply(args *base.TestingKnobs) {
+	e := args.SQLEvalContext.(*eval.TestingKnobs)
+	v := true
+	e.UnsafeOverride = func() *bool { return &v }
+}
+
 // clusterOptIgnoreStrictGCForTenants corresponds to the
 // ignore-tenant-strict-gc-enforcement directive.
 type clusterOptIgnoreStrictGCForTenants struct{}
@@ -2268,6 +2289,8 @@ func readKnobOptions(t *testing.T, path string) []knobOpt {
 			res = append(res, knobOptDisableCorpusGeneration{})
 		case "sync-event-log":
 			res = append(res, knobOptSynchronousEventLog{})
+		case "allow-unsafe":
+			res = append(res, knobOptAllowUnsafe{})
 		default:
 			t.Fatalf("unrecognized knob option: %s", opt)
 		}
@@ -2954,6 +2977,9 @@ func (t *logicTest) processSubtest(
 						case "async":
 							query.expectAsync = true
 
+						case "allowunsafe":
+							query.allowUnsafe = true
+
 						default:
 							if strings.HasPrefix(opt, "round-in-strings") {
 								significantFigures, err := floatcmp.ParseRoundInStringsDirective(opt)
@@ -3599,6 +3625,7 @@ func (t *logicTest) unexpectedError(sql string, pos string, err error) (bool, er
 var uniqueHashPattern = regexp.MustCompile(`UNIQUE.*USING\s+HASH`)
 
 func (t *logicTest) execStatement(stmt logicStatement, disableCFMutator bool) (bool, error) {
+	defer t.setSafetyGate(stmt.sql, false)()
 	db := t.db
 	t.noticeBuffer = nil
 	if *showSQL {
@@ -3722,6 +3749,7 @@ func (t *logicTest) hashResults(results []string) (string, error) {
 }
 
 func (t *logicTest) execQuery(query logicQuery) error {
+	defer t.setSafetyGate(query.sql, query.allowUnsafe)()
 	if *showSQL {
 		t.outf("%s;", query.sql)
 	}
@@ -4609,6 +4637,7 @@ func RunLogicTest(
 		perErrorSummary:            make(map[string][]string),
 		rng:                        rng,
 		declarativeCorpusCollector: cc,
+		allowUnsafe:                true,
 	}
 	if *printErrorSummary {
 		defer lt.printErrorSummary()
@@ -4884,3 +4913,23 @@ func locateCockroachPredecessor(version string) (string, error) {
 	}
 	return path, nil
 }
+
+// setSafetyGate is a utility function which controls whether access to the unsafe
+// internals is allowed by the contained sql statement. The reasoning behind it is
+// as follows. We want queries which explicitly access unsafe internals to have
+// access to them, but we want to prevent indirect access wherever possible.
+// Indirect access can be described as any query which doesn't reference an unsafe
+// object, but still accesses it under the hood. We want to flush out these cases
+// so that users can never execute statements which look safe, but block when executed.
+func (t *logicTest) setSafetyGate(sql string, skip bool) func() {
+	sql = strings.ToLower(sql)
+	explicitlyUnsafe := strings.Contains(sql, "crdb_internal.") || strings.Contains(sql, "system.")
+	if skip || explicitlyUnsafe {
+		return func() {}
+	}
+
+	t.allowUnsafe = false
+	return func() {
+		t.allowUnsafe = true
+	}
+}

pkg/sql/logictest/parallel_test.go

@@ -67,13 +67,14 @@ func (t *parallelTest) processTestFile(path string, nodeIdx int, db *gosql.DB, c
 	// Set up a dummy logicTest structure to use that code.
 	rng, _ := randutil.NewTestRand()
 	l := &logicTest{
-		rootT:   t.T,
-		cluster: t.cluster,
-		nodeIdx: nodeIdx,
-		db:      db,
-		user:    username.RootUser,
-		verbose: testing.Verbose() || log.V(1),
-		rng:     rng,
+		rootT:       t.T,
+		cluster:     t.cluster,
+		nodeIdx:     nodeIdx,
+		db:          db,
+		user:        username.RootUser,
+		verbose:     testing.Verbose() || log.V(1),
+		rng:         rng,
+		allowUnsafe: true,
 	}
 	if err := l.processTestFile(path, logictest.TestClusterConfig{}); err != nil {
 		log.Dev.Errorf(context.Background(), "error processing %s: %s", path, err)

pkg/sql/logictest/testdata/logic_test/bytes

@@ -156,7 +156,7 @@ PREPARE r1(bytes) AS
 		SELECT id FROM system.namespace WHERE name = 'defaultdb'
 	);
 
-query T
+query T allowunsafe
 EXECUTE r1('abc')
 ----
 \022[\012\011defaultdb\020d\0320\012\013\012\005admin\020\002\030\002\012\015\012\006public\020\200\020\030\000\012\012\012\004root\020\002\030\002\022\004root\030\003"\000(\001:\014\012\006public\022\002\010e@\000J\000Z\002\020\000p\000