Merge #135149

135149: scbuild: fix unhandled case during DROP SCHEMA r=rafiss a=rafiss fixes #135141 fixes #133913 fixes #134752 fixes #134280 fixes #133510 Release note (bug fix): Fixed an unhandled error that would occur if DROP SCHEMA was executed on the `public` schema of the `system` database, or on an internal schema like `pg_catalog` or `information_schema`. Co-authored-by: Rafi Shamim <[email protected]>
cockroachdb · Nov 14, 2024 · 724c140 · 724c140
2 parents 36d4b83 + 36e0144
commit 724c140
Show file tree

Hide file tree

Showing 3 changed files with 40 additions and 11 deletions.
diff --git a/pkg/sql/drop_schema.go b/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 {

diff --git a/pkg/sql/logictest/testdata/logic_test/drop_schema b/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
diff --git a/pkg/sql/schemachanger/scbuild/builder_state.go b/pkg/sql/schemachanger/scbuild/builder_state.go
@@ -309,7 +309,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())))
 	}
 }
 
@@ -989,8 +989,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 {