sql: support the USAGE privilege on schemas

pkg/sql/authorization.go

@@ -515,6 +515,31 @@ func (p *planner) canCreateOnSchema(ctx context.Context, schemaID descpb.ID, use
 	}
 }
 
+func (p *planner) canResolveDescUnderSchema(
+	ctx context.Context, schemaID descpb.ID, desc catalog.Descriptor,
+) error {
+	// We can't always resolve temporary schemas by ID (for example in the temporary
+	// object cleaner which accesses temporary schemas not in the current session).
+	// To avoid an internal error, we just don't check usage on temporary tables.
+	if tbl, ok := desc.(catalog.TableDescriptor); ok && tbl.IsTemporary() {
+		return nil
+	}
+	resolvedSchema, err := p.Descriptors().ResolveSchemaByID(ctx, p.Txn(), schemaID)
+	if err != nil {
+		return err
+	}
+
+	switch resolvedSchema.Kind {
+	case catalog.SchemaPublic, catalog.SchemaTemporary, catalog.SchemaVirtual:
+		// Anyone can resolve under temporary, public or virtual schemas.
+		return nil
+	case catalog.SchemaUserDefined:
+		return p.CheckPrivilegeForUser(ctx, resolvedSchema.Desc, privilege.USAGE, p.User())
+	default:
+		panic(errors.AssertionFailedf("unknown schema kind %d", resolvedSchema.Kind))
+	}
+}
+
 // checkCanAlterToNewOwner checks if the new owner exists, the current user
 // has privileges to alter the owner of the object and the current user is a
 //  member of the new owner role.

pkg/sql/logictest/testdata/logic_test/schema

@@ -320,4 +320,35 @@ ALTER SCHEMA privs RENAME TO denied
 statement error pq: permission denied to drop schema "privs"
 DROP SCHEMA privs
 
-# TODO (rohany): Test the usage privilege here.
+# Test the usage privilege.
+user root
+
+# Create some objects in privs (testuser doesn't have USAGE yet).
+statement ok
+CREATE TABLE privs.usage_tbl (x INT);
+CREATE TYPE privs.usage_typ AS ENUM ('usage');
+
+user testuser
+
+# Both mutable and immutable access should fail with this error.
+
+statement error pq: user testuser does not have USAGE privilege on schema privs
+SELECT * FROM privs.usage_tbl
+
+statement error pq: user testuser does not have USAGE privilege on schema privs
+SELECT 'usage'::privs.usage_typ
+
+statement error pq: user testuser does not have USAGE privilege on schema privs
+ALTER TABLE privs.usage_tbl ADD COLUMN y INT DEFAULT NULL
+
+statement error pq: user testuser does not have USAGE privilege on schema privs
+CREATE INDEX ON privs.usage_tbl (x)
+
+statement error pq: user testuser does not have USAGE privilege on schema privs
+COMMENT ON TABLE privs.usage_tbl IS 'foo'
+
+statement error pq: user testuser does not have USAGE privilege on schema privs
+COMMENT ON COLUMN privs.usage_tbl.x IS 'foo'
+
+statement error pq: user testuser does not have USAGE privilege on schema privs
+ALTER TYPE privs.usage_typ ADD VALUE 'denied'

pkg/sql/opt_catalog.go

@@ -207,6 +207,12 @@ func (oc *optCatalog) ResolveDataSource(
 	if err != nil {
 		return nil, cat.DataSourceName{}, err
 	}
+
+	// Ensure that the current user can access the target schema.
+	if err := oc.planner.canResolveDescUnderSchema(ctx, desc.GetParentSchemaID(), desc); err != nil {
+		return nil, cat.DataSourceName{}, err
+	}
+
 	ds, err := oc.dataSourceForDesc(ctx, flags, desc, &oc.tn)
 	if err != nil {
 		return nil, cat.DataSourceName{}, err

pkg/sql/resolver.go

@@ -128,7 +128,19 @@ type resolveFlags struct {
 func (p *planner) ResolveMutableTableDescriptor(
 	ctx context.Context, tn *tree.TableName, required bool, requiredType tree.RequiredTableKind,
 ) (table *tabledesc.Mutable, err error) {
-	return resolver.ResolveMutableExistingTableObject(ctx, p, tn, required, requiredType)
+	desc, err := resolver.ResolveMutableExistingTableObject(ctx, p, tn, required, requiredType)
+	if err != nil {
+		return nil, err
+	}
+
+	// Ensure that the current user can access the target schema.
+	if desc != nil {
+		if err := p.canResolveDescUnderSchema(ctx, desc.GetParentSchemaID(), desc); err != nil {
+			return nil, err
+		}
+	}
+
+	return desc, nil
 }
 
 func (p *planner) ResolveUncachedTableDescriptor(
@@ -142,7 +154,18 @@ func (p *planner) ResolveUncachedTableDescriptor(
 		}
 		table, err = resolver.ResolveExistingTableObject(ctx, p, tn, lookupFlags)
 	})
-	return table, err
+	if err != nil {
+		return nil, err
+	}
+
+	// Ensure that the current user can access the target schema.
+	if table != nil {
+		if err := p.canResolveDescUnderSchema(ctx, table.GetParentSchemaID(), table); err != nil {
+			return nil, err
+		}
+	}
+
+	return table, nil
 }
 
 func (p *planner) ResolveTargetObject(
@@ -249,6 +272,11 @@ func (p *planner) ResolveType(
 			pgcode.FeatureNotSupported, "cross database type references are not supported: %s", tn.String())
 	}
 
+	// Ensure that the user can access the target schema.
+	if err := p.canResolveDescUnderSchema(ctx, tdesc.GetParentSchemaID(), tdesc); err != nil {
+		return nil, err
+	}
+
 	return tdesc.MakeTypesT(ctx, &tn, p)
 }
 
@@ -790,6 +818,14 @@ func (p *planner) ResolveMutableTypeDescriptor(
 		return nil, err
 	}
 	name.SetAnnotation(&p.semaCtx.Annotations, tn)
+
+	if desc != nil {
+		// Ensure that the user can access the target schema.
+		if err := p.canResolveDescUnderSchema(ctx, desc.GetParentSchemaID(), desc); err != nil {
+			return nil, err
+		}
+	}
+
 	return desc, nil
 }
 
@@ -809,6 +845,14 @@ func (p *planner) ResolveMutableTableDescriptorEx(
 		return nil, err
 	}
 	name.SetAnnotation(&p.semaCtx.Annotations, &tn)
+
+	if table != nil {
+		// Ensure that the user can access the target schema.
+		if err := p.canResolveDescUnderSchema(ctx, table.GetParentSchemaID(), table); err != nil {
+			return nil, err
+		}
+	}
+
 	return table, nil
 }
 
@@ -834,7 +878,14 @@ func (p *planner) ResolveMutableTableDescriptorExAllowNoPrimaryKey(
 	}
 	tn := tree.MakeTableNameFromPrefix(prefix, tree.Name(name.Object()))
 	name.SetAnnotation(&p.semaCtx.Annotations, &tn)
-	return desc.(*tabledesc.Mutable), nil
+	table := desc.(*tabledesc.Mutable)
+
+	// Ensure that the user can access the target schema.
+	if err := p.canResolveDescUnderSchema(ctx, table.GetParentSchemaID(), table); err != nil {
+		return nil, err
+	}
+
+	return table, nil
 }
 
 // See ResolveUncachedTableDescriptor.
@@ -847,7 +898,18 @@ func (p *planner) ResolveUncachedTableDescriptorEx(
 	p.runWithOptions(resolveFlags{skipCache: true}, func() {
 		table, err = p.ResolveExistingObjectEx(ctx, name, required, requiredType)
 	})
-	return table, err
+	if err != nil {
+		return nil, err
+	}
+
+	if table != nil {
+		// Ensure that the user can access the target schema.
+		if err := p.canResolveDescUnderSchema(ctx, table.GetParentSchemaID(), table); err != nil {
+			return nil, err
+		}
+	}
+
+	return table, nil
 }
 
 // See ResolveExistingTableObject.
@@ -868,7 +930,14 @@ func (p *planner) ResolveExistingObjectEx(
 	}
 	tn := tree.MakeTableNameFromPrefix(prefix, tree.Name(name.Object()))
 	name.SetAnnotation(&p.semaCtx.Annotations, &tn)
-	return desc.(*tabledesc.Immutable), nil
+	table := desc.(*tabledesc.Immutable)
+
+	// Ensure that the user can access the target schema.
+	if err := p.canResolveDescUnderSchema(ctx, table.GetParentSchemaID(), table); err != nil {
+		return nil, err
+	}
+
+	return table, nil
 }
 
 // ResolvedName is a convenience wrapper for UnresolvedObjectName.Resolved.