Merge pull request #135181 from cockroachdb/blathers/backport-release…

…-24.3-135149

release-24.3: scbuild: fix unhandled case during DROP SCHEMA

pkg/sql/drop_schema.go

@@ -70,6 +70,14 @@ func (p *planner) DropSchema(ctx context.Context, n *tree.DropSchema) (planNode,
 			}
 			return nil, pgerror.Newf(pgcode.InvalidSchemaName, "unknown schema %q", scName)
 		}
+		hasOwnership, err := p.HasOwnership(ctx, sc)
+		if err != nil {
+			return nil, err
+		}
+		if !hasOwnership {
+			return nil, pgerror.Newf(pgcode.InsufficientPrivilege,
+				"must be owner of schema %s", tree.Name(sc.GetName()))
+		}
 
 		if scName == catconstants.PublicSchemaName {
 			return nil, pgerror.Newf(pgcode.InvalidSchemaName, "cannot drop schema %q", scName)
@@ -79,14 +87,6 @@ func (p *planner) DropSchema(ctx context.Context, n *tree.DropSchema) (planNode,
 		case catalog.SchemaPublic, catalog.SchemaVirtual, catalog.SchemaTemporary:
 			return nil, pgerror.Newf(pgcode.InvalidSchemaName, "cannot drop schema %q", scName)
 		case catalog.SchemaUserDefined:
-			hasOwnership, err := p.HasOwnership(ctx, sc)
-			if err != nil {
-				return nil, err
-			}
-			if !hasOwnership {
-				return nil, pgerror.Newf(pgcode.InsufficientPrivilege,
-					"must be owner of schema %s", tree.Name(sc.GetName()))
-			}
 			namesBefore := len(d.objectNamesToDelete)
 			fnsBefore := len(d.functionsToDelete)
 			if err := d.collectObjectsInSchema(ctx, p, db, sc); err != nil {

pkg/sql/logictest/testdata/logic_test/drop_schema

@@ -21,3 +21,22 @@ BEGIN;
 ALTER TYPE schema_123539.enum_123539 DROP VALUE 's';
 DROP SCHEMA schema_123539 CASCADE;
 COMMIT;
+
+# Check that we block dropping the public schema of the system database, as
+# well as virtual schemas.
+
+statement error must be owner of schema public
+DROP SCHEMA system.public
+
+statement error must be owner of schema pg_catalog
+DROP SCHEMA pg_catalog
+
+user testuser
+
+statement error must be owner of schema public
+DROP SCHEMA system.public
+
+statement error must be owner of schema crdb_internal
+DROP SCHEMA crdb_internal
+
+user root

pkg/sql/schemachanger/scbuild/builder_state.go

@@ -311,7 +311,7 @@ func (b *builderState) mustOwn(id catid.DescID) {
 	b.ensureDescriptor(id)
 	if c := b.descCache[id]; !c.hasOwnership {
 		panic(pgerror.Newf(pgcode.InsufficientPrivilege,
-			"must be owner of %s %s", c.desc.DescriptorType(), c.desc.GetName()))
+			"must be owner of %s %s", c.desc.DescriptorType(), tree.Name(c.desc.GetName())))
 	}
 }
 
@@ -991,8 +991,18 @@ func (b *builderState) checkOwnershipOrPrivilegesOnSchemaDesc(
 	case catalog.SchemaTemporary:
 		// Nothing needs to be done.
 	case catalog.SchemaPublic, catalog.SchemaVirtual:
-		panic(pgerror.Newf(pgcode.InsufficientPrivilege,
-			"%s permission denied for schema %q", p.RequiredPrivilege.DisplayName(), name))
+		if p.RequireOwnership {
+			if ok, err := b.auth.HasOwnership(b.ctx, sc); err != nil {
+				panic(err)
+			} else if !ok {
+				panic(pgerror.Newf(pgcode.InsufficientPrivilege,
+					"must be owner of schema %s", tree.Name(name.Schema())))
+			}
+		} else {
+			if err := b.auth.CheckPrivilege(b.ctx, sc, p.RequiredPrivilege); err != nil {
+				panic(err)
+			}
+		}
 	case catalog.SchemaUserDefined:
 		b.ensureDescriptor(sc.GetID())
 		if p.RequireOwnership {