sql: add setting to control CREATE privilege on public schema

docs/generated/settings/settings-for-tenants.txt

@@ -89,6 +89,7 @@ server.user_login.upgrade_bcrypt_stored_passwords_to_scram.enabled	boolean	true
 server.web_session.purge.ttl	duration	1h0m0s	if nonzero, entries in system.web_sessions older than this duration are periodically purged	tenant-rw
 server.web_session_timeout	duration	168h0m0s	the duration that a newly created web session will be valid	tenant-rw
 sql.auth.change_own_password.enabled	boolean	false	controls whether a user is allowed to change their own password, even if they have no other privileges	tenant-rw
+sql.auth.public_schema_create_privilege.enabled	boolean	true	determines whether to grant all users the CREATE privileges on the public schema when it is created	tenant-rw
 sql.auth.resolve_membership_single_scan.enabled	boolean	true	determines whether to populate the role membership cache with a single scan	tenant-rw
 sql.closed_session_cache.capacity	integer	1000	the maximum number of sessions in the cache	tenant-rw
 sql.closed_session_cache.time_to_live	integer	3600	the maximum time to live, in seconds	tenant-rw

docs/generated/settings/settings.html

@@ -123,6 +123,7 @@
 <tr><td><div id="setting-spanconfig-storage-coalesce-adjacent-enabled" class="anchored"><code>spanconfig.storage_coalesce_adjacent.enabled</code></div></td><td>boolean</td><td><code>true</code></td><td>collapse adjacent ranges with the same span configs, for the ranges specific to the system tenant</td><td>Dedicated/Self-Hosted</td></tr>
 <tr><td><div id="setting-spanconfig-tenant-coalesce-adjacent-enabled" class="anchored"><code>spanconfig.tenant_coalesce_adjacent.enabled</code></div></td><td>boolean</td><td><code>true</code></td><td>collapse adjacent ranges with the same span configs across all secondary tenant keyspaces</td><td>Dedicated/Self-Hosted</td></tr>
 <tr><td><div id="setting-sql-auth-change-own-password-enabled" class="anchored"><code>sql.auth.change_own_password.enabled</code></div></td><td>boolean</td><td><code>false</code></td><td>controls whether a user is allowed to change their own password, even if they have no other privileges</td><td>Serverless/Dedicated/Self-Hosted</td></tr>
+<tr><td><div id="setting-sql-auth-public-schema-create-privilege-enabled" class="anchored"><code>sql.auth.public_schema_create_privilege.enabled</code></div></td><td>boolean</td><td><code>true</code></td><td>determines whether to grant all users the CREATE privileges on the public schema when it is created</td><td>Serverless/Dedicated/Self-Hosted</td></tr>
 <tr><td><div id="setting-sql-auth-resolve-membership-single-scan-enabled" class="anchored"><code>sql.auth.resolve_membership_single_scan.enabled</code></div></td><td>boolean</td><td><code>true</code></td><td>determines whether to populate the role membership cache with a single scan</td><td>Serverless/Dedicated/Self-Hosted</td></tr>
 <tr><td><div id="setting-sql-closed-session-cache-capacity" class="anchored"><code>sql.closed_session_cache.capacity</code></div></td><td>integer</td><td><code>1000</code></td><td>the maximum number of sessions in the cache</td><td>Serverless/Dedicated/Self-Hosted</td></tr>
 <tr><td><div id="setting-sql-closed-session-cache-time-to-live" class="anchored"><code>sql.closed_session_cache.time_to_live</code></div></td><td>integer</td><td><code>3600</code></td><td>the maximum time to live, in seconds</td><td>Serverless/Dedicated/Self-Hosted</td></tr>

pkg/ccl/backupccl/restore_job.go

@@ -1074,10 +1074,10 @@ func createImportingDescriptors(
 			}
 			descsCol := txn.Descriptors()
 			// Write the new descriptors which are set in the OFFLINE state.
+			includePublicSchemaCreatePriv := sql.PublicSchemaCreatePrivilegeEnabled.Get(&p.ExecCfg().Settings.SV)
 			if err := ingesting.WriteDescriptors(
-				ctx, p.ExecCfg().Codec, txn.KV(), p.User(), descsCol,
-				databases, writtenSchemas, tables, writtenTypes, writtenFunctions,
-				details.DescriptorCoverage, nil /* extra */, restoreTempSystemDB,
+				ctx, txn.KV(), p.User(), descsCol, databases, writtenSchemas, tables, writtenTypes, writtenFunctions,
+				details.DescriptorCoverage, nil /* extra */, restoreTempSystemDB, includePublicSchemaCreatePriv,
 			); err != nil {
 				return errors.Wrapf(err, "restoring %d TableDescriptors from %d databases", len(tables), len(databases))
 			}
@@ -1461,10 +1461,12 @@ func remapPublicSchemas(
 			return err
 		}
 
+		includeCreatePriv := sql.PublicSchemaCreatePrivilegeEnabled.Get(p.ExecCfg().SV())
+
 		db.AddSchemaToDatabase(tree.PublicSchema, descpb.DatabaseDescriptor_SchemaInfo{ID: id})
 		// Every database must be initialized with the public schema.
 		// Create the SchemaDescriptor.
-		publicSchemaPrivileges := catpb.NewPublicSchemaPrivilegeDescriptor()
+		publicSchemaPrivileges := catpb.NewPublicSchemaPrivilegeDescriptor(includeCreatePriv)
 		publicSchemaDesc := schemadesc.NewBuilder(&descpb.SchemaDescriptor{
 			ParentID:   db.GetID(),
 			Name:       tree.PublicSchema,

pkg/sql/catalog/catpb/privilege.go

@@ -189,15 +189,17 @@ func NewBaseDatabasePrivilegeDescriptor(owner username.SQLUsername) *PrivilegeDe
 // descriptor owned by the admin user which has CREATE and USAGE privilege for
 // the public role, and ALL privileges for superusers. It is used for the
 // public schema.
-func NewPublicSchemaPrivilegeDescriptor() *PrivilegeDescriptor {
+func NewPublicSchemaPrivilegeDescriptor(includeCreatePriv bool) *PrivilegeDescriptor {
 	// In postgres, the user "postgres" is the owner of the public schema in a
 	// newly created db. In CockroachDB, admin is our substitute for the postgres
 	// user.
 	p := NewBasePrivilegeDescriptor(username.AdminRoleName())
-	// By default, everyone has USAGE and CREATE on the public schema.
-	// Once https://github.com/cockroachdb/cockroach/issues/70266 is resolved,
-	// the public role will no longer have CREATE privileges.
-	p.Grant(username.PublicRoleName(), privilege.List{privilege.CREATE, privilege.USAGE}, false)
+
+	if includeCreatePriv {
+		p.Grant(username.PublicRoleName(), privilege.List{privilege.CREATE, privilege.USAGE}, false)
+	} else {
+		p.Grant(username.PublicRoleName(), privilege.List{privilege.USAGE}, false)
+	}
 	return p
 }
 

pkg/sql/catalog/ingesting/privileges.go

@@ -45,6 +45,7 @@ func GetIngestingDescriptorPrivileges(
 	wroteDBs map[descpb.ID]catalog.DatabaseDescriptor,
 	wroteSchemas map[descpb.ID]catalog.SchemaDescriptor,
 	descCoverage tree.DescriptorCoverage,
+	includePublicSchemaCreatePriv bool,
 ) (updatedPrivileges *catpb.PrivilegeDescriptor, err error) {
 	switch desc := desc.(type) {
 	case catalog.TableDescriptor:
@@ -58,6 +59,7 @@ func GetIngestingDescriptorPrivileges(
 			wroteSchemas,
 			descCoverage,
 			privilege.Table,
+			includePublicSchemaCreatePriv,
 		)
 	case catalog.SchemaDescriptor:
 		return getIngestingPrivilegesForTableOrSchema(
@@ -70,6 +72,7 @@ func GetIngestingDescriptorPrivileges(
 			wroteSchemas,
 			descCoverage,
 			privilege.Schema,
+			includePublicSchemaCreatePriv,
 		)
 	case catalog.TypeDescriptor:
 		// If the ingestion is not a cluster restore we cannot know that the users
@@ -103,6 +106,7 @@ func getIngestingPrivilegesForTableOrSchema(
 	wroteSchemas map[descpb.ID]catalog.SchemaDescriptor,
 	descCoverage tree.DescriptorCoverage,
 	privilegeType privilege.ObjectType,
+	includePublicSchemaCreatePriv bool,
 ) (updatedPrivileges *catpb.PrivilegeDescriptor, err error) {
 	if _, ok := wroteDBs[desc.GetParentID()]; ok {
 		// If we're creating a new database in this ingestion, the tables and
@@ -111,7 +115,7 @@ func getIngestingPrivilegesForTableOrSchema(
 		switch privilegeType {
 		case privilege.Schema:
 			if desc.GetName() == tree.PublicSchema {
-				updatedPrivileges = catpb.NewPublicSchemaPrivilegeDescriptor()
+				updatedPrivileges = catpb.NewPublicSchemaPrivilegeDescriptor(includePublicSchemaCreatePriv)
 			} else {
 				updatedPrivileges = catpb.NewBasePrivilegeDescriptor(user)
 			}

pkg/sql/catalog/ingesting/write_descs.go

@@ -13,7 +13,6 @@ package ingesting
 import (
 	"context"
 
-	"github.com/cockroachdb/cockroach/pkg/keys"
 	"github.com/cockroachdb/cockroach/pkg/kv"
 	"github.com/cockroachdb/cockroach/pkg/kv/kvpb"
 	"github.com/cockroachdb/cockroach/pkg/roachpb"
@@ -46,7 +45,6 @@ import (
 // inherited privileges during a cluster restore.
 func WriteDescriptors(
 	ctx context.Context,
-	codec keys.SQLCodec,
 	txn *kv.Txn,
 	user username.SQLUsername,
 	descsCol *descs.Collection,
@@ -58,6 +56,7 @@ func WriteDescriptors(
 	descCoverage tree.DescriptorCoverage,
 	extra []roachpb.KeyValue,
 	inheritParentName string,
+	includePublicSchemaCreatePriv bool,
 ) (err error) {
 	ctx, span := tracing.ChildSpan(ctx, "WriteDescriptors")
 	defer span.Finish()
@@ -80,7 +79,7 @@ func WriteDescriptors(
 	for i := range databases {
 		desc := databases[i]
 		updatedPrivileges, err := GetIngestingDescriptorPrivileges(ctx, txn, descsCol, desc, user,
-			wroteDBs, wroteSchemas, descCoverage)
+			wroteDBs, wroteSchemas, descCoverage, includePublicSchemaCreatePriv)
 		if err != nil {
 			return err
 		}
@@ -119,7 +118,7 @@ func WriteDescriptors(
 	for i := range schemas {
 		sc := schemas[i]
 		updatedPrivileges, err := GetIngestingDescriptorPrivileges(ctx, txn, descsCol, sc, user,
-			wroteDBs, wroteSchemas, descCoverage)
+			wroteDBs, wroteSchemas, descCoverage, includePublicSchemaCreatePriv)
 		if err != nil {
 			return err
 		}
@@ -147,7 +146,7 @@ func WriteDescriptors(
 	for i := range tables {
 		table := tables[i]
 		updatedPrivileges, err := GetIngestingDescriptorPrivileges(ctx, txn, descsCol, table, user,
-			wroteDBs, wroteSchemas, descCoverage)
+			wroteDBs, wroteSchemas, descCoverage, includePublicSchemaCreatePriv)
 		if err != nil {
 			return err
 		}
@@ -181,7 +180,7 @@ func WriteDescriptors(
 	for i := range types {
 		typ := types[i]
 		updatedPrivileges, err := GetIngestingDescriptorPrivileges(ctx, txn, descsCol, typ, user,
-			wroteDBs, wroteSchemas, descCoverage)
+			wroteDBs, wroteSchemas, descCoverage, includePublicSchemaCreatePriv)
 		if err != nil {
 			return err
 		}
@@ -205,7 +204,7 @@ func WriteDescriptors(
 
 	for _, fn := range functions {
 		updatedPrivileges, err := GetIngestingDescriptorPrivileges(
-			ctx, txn, descsCol, fn, user, wroteDBs, wroteSchemas, descCoverage,
+			ctx, txn, descsCol, fn, user, wroteDBs, wroteSchemas, descCoverage, includePublicSchemaCreatePriv,
 		)
 		if err != nil {
 			return err

pkg/sql/catalog/randgen/randgen.go

@@ -53,7 +53,7 @@ func NewTestSchemaGenerator(
 ) TestSchemaGenerator {
 	dbPrivs := catpb.NewBaseDatabasePrivilegeDescriptor(callingUser)
 	dbDefaultPrivs := catprivilege.MakeDefaultPrivilegeDescriptor(catpb.DefaultPrivilegeDescriptor_DATABASE)
-	publicSchemaPrivs := catpb.NewPublicSchemaPrivilegeDescriptor()
+	publicSchemaPrivs := catpb.NewPublicSchemaPrivilegeDescriptor(true /*includeCreatePriv*/)
 	publicSchemaPrivs.SetOwner(callingUser)
 	g := &testSchemaGenerator{
 		rand: rand,

pkg/sql/catalog/schemadesc/public_schema_desc.go

@@ -55,7 +55,7 @@ var _ syntheticBase = publicBase{}
 func (publicBase) kindName() string                 { return "public" }
 func (publicBase) kind() catalog.ResolvedSchemaKind { return catalog.SchemaPublic }
 func (publicBase) GetPrivileges() *catpb.PrivilegeDescriptor {
-	return catpb.NewPublicSchemaPrivilegeDescriptor()
+	return catpb.NewPublicSchemaPrivilegeDescriptor(true /*includeCreatePriv*/)
 }
 
 // publicDesc is a singleton returned by GetPublicSchema.

pkg/sql/descriptor.go

@@ -59,6 +59,17 @@ var (
 	errNoMatch           = pgerror.New(pgcode.UndefinedObject, "no object matched")
 )
 
+// PublicSchemaCreatePrivilegeEnabled is the cluster setting that determines
+// whether the CREATE privilege is given to the `public` role on the `public`
+// schema at the time the schema is created.
+var PublicSchemaCreatePrivilegeEnabled = settings.RegisterBoolSetting(
+	settings.TenantWritable,
+	"sql.auth.public_schema_create_privilege.enabled",
+	"determines whether to grant all users the CREATE privileges on the public "+
+		"schema when it is created",
+	true,
+).WithPublic()
+
 // createDatabase takes Database descriptor and creates it if needed,
 // incrementing the descriptor counter. Returns true if the descriptor
 // is actually created, false if it already existed, or an error if one was
@@ -145,11 +156,12 @@ func (p *planner) createDatabase(
 		dbdesc.MaybeWithDatabaseRegionConfig(regionConfig),
 		dbdesc.WithPublicSchemaID(publicSchemaID),
 	)
+	includeCreatePriv := PublicSchemaCreatePrivilegeEnabled.Get(&p.execCfg.Settings.SV)
 	publicSchema := schemadesc.NewBuilder(&descpb.SchemaDescriptor{
 		ParentID:   id,
 		Name:       tree.PublicSchema,
 		ID:         publicSchemaID,
-		Privileges: catpb.NewPublicSchemaPrivilegeDescriptor(),
+		Privileges: catpb.NewPublicSchemaPrivilegeDescriptor(includeCreatePriv),
 		Version:    1,
 	}).BuildCreatedMutableSchema()
 

pkg/sql/importer/import_job.go

@@ -600,10 +600,12 @@ func prepareNewTablesForIngestion(
 	// Write the new TableDescriptors and flip the namespace entries over to
 	// them. After this call, any queries on a table will be served by the newly
 	// imported data.
+	includePublicSchemaCreatePriv := sql.PublicSchemaCreatePrivilegeEnabled.Get(&p.ExecCfg().Settings.SV)
 	if err := ingesting.WriteDescriptors(
-		ctx, p.ExecCfg().Codec, txn, p.User(), descsCol,
-		nil /* databases */, nil /* schemas */, tableDescs, nil /* types */, nil, /* functions */
-		tree.RequestedDescriptors, seqValKVs, "" /* inheritParentName */); err != nil {
+		ctx, txn, p.User(), descsCol, nil /* databases */, nil /* schemas */, tableDescs,
+		nil /* types */, nil /* functions */, tree.RequestedDescriptors, seqValKVs,
+		"" /* inheritParentName */, includePublicSchemaCreatePriv,
+	); err != nil {
 		return nil, errors.Wrapf(err, "creating importTables")
 	}
 

pkg/sql/logictest/testdata/logic_test/schema

@@ -1324,3 +1324,33 @@ DROP DATABASE testdb1 CASCADE;
 DROP DATABASE testdb2 CASCADE;
 
 subtest end
+
+subtest public_schema_create_privilege
+
+statement ok
+CREATE DATABASE should_have_create
+
+statement ok
+USE should_have_create
+
+query TTTT
+SELECT database_name, schema_name, grantee, privilege_type FROM [SHOW GRANTS ON SCHEMA public] WHERE grantee = 'public'
+----
+should_have_create  public  public  CREATE
+should_have_create  public  public  USAGE
+
+statement ok
+SET CLUSTER SETTING sql.auth.public_schema_create_privilege.enabled = false
+
+statement ok
+CREATE DATABASE should_not_have_create
+
+statement ok
+USE should_not_have_create
+
+query TTTT
+SELECT database_name, schema_name, grantee, privilege_type FROM [SHOW GRANTS ON SCHEMA public] WHERE grantee = 'public'
+----
+should_not_have_create  public  public  USAGE
+
+subtest end

pkg/sql/logictest/testdata/logic_test/schema_locked

@@ -113,5 +113,3 @@ statement ok
 DROP TABLE t;
 
 
-
-