sql: add IsSuperUser for proper error handling [with help of @mgm702]

Release note: None

docs/RFCS/20171220_sql_role_based_access_control.md

@@ -439,8 +439,10 @@ Role membership expansion can be pre-computed (see [Internal representation of m
 Permission checks currently consist of calls to one of:
 ```
 func (p *planner) CheckPrivilege(descriptor sqlbase.DescriptorProto, privilege privilege.Kind) error
+func (p *planner) CheckAnyPrivilege(descriptor sqlbase.DescriptorProto) error
 func (p *planner) RequireSuperUser(action string) error
-func (p *planner) anyPrivilege(descriptor sqlbase.DescriptorProto) error
+func (p *planner) IsSuperUser(action string) (bool, error)
+func (p *planner) MemberOfWithAdminOption(member string) (map[string]bool, error)
 ```
 
 All methods operate on `p.session.User`.

pkg/ccl/roleccl/role.go

@@ -72,7 +72,9 @@ func grantRolePlanHook(
 		return nil, err
 	}
 
-	if err := p.RequireSuperUser(ctx, "grant role"); err != nil {
+	if isSuperUser, err := p.IsSuperUser(ctx, "grant role"); err != nil {
+		return nil, err
+	} else if !isSuperUser {
 		// Not a superuser: check permissions on each role.
 		allRoles, err := p.MemberOfWithAdminOption(ctx, p.User())
 		if err != nil {
@@ -201,7 +203,9 @@ func revokeRolePlanHook(
 		return nil, err
 	}
 
-	if err := p.RequireSuperUser(ctx, "revoke role"); err != nil {
+	if isSuperUser, err := p.IsSuperUser(ctx, "revoke role"); err != nil {
+		return nil, err
+	} else if !isSuperUser {
 		// Not a superuser: check permissions on each role.
 		allRoles, err := p.MemberOfWithAdminOption(ctx, p.User())
 		if err != nil {

pkg/sql/authorization.go

@@ -46,8 +46,16 @@ type AuthorizationAccessor interface {
 	// CheckAnyPrivilege returns nil if user has any privileges at all.
 	CheckAnyPrivilege(ctx context.Context, descriptor sqlbase.DescriptorProto) error
 
-	// RequiresSuperUser errors if the session user isn't a super-user (i.e. root
-	// or node). Includes the named action in the error message.
+	// IsSuperUser returns tuple of bool and error:
+	// (true, nil) means that the user is a superuser (i.e. root or node)
+	// (false, nil) means that the user is NOT a superuser
+	// (false, err) means that there was an error running the query on
+	// the `system.users` table
+	IsSuperUser(ctx context.Context, action string) (bool, error)
+
+	// RequiresSuperUser is a wrapper on top of IsSuperUser.
+	// It errors if IsSuperUser errors or if the user isn't a super-user.
+	// Includes the named action in the error message.
 	RequireSuperUser(ctx context.Context, action string) error
 
 	// MemberOfWithAdminOption looks up all the roles (direct and indirect) that 'member' is a member
@@ -145,28 +153,42 @@ func (p *planner) CheckAnyPrivilege(ctx context.Context, descriptor sqlbase.Desc
 		p.SessionData().User, descriptor.TypeName(), descriptor.GetName())
 }
 
-// RequireSuperUser implements the AuthorizationAccessor interface.
-func (p *planner) RequireSuperUser(ctx context.Context, action string) error {
+// IsSuperUser implements the AuthorizationAccessor interface.
+func (p *planner) IsSuperUser(ctx context.Context, action string) (bool, error) {
 	user := p.SessionData().User
 
 	// Check if user is 'root' or 'node'.
 	if user == security.RootUser || user == security.NodeUser {
-		return nil
+		return true, nil
 	}
 
 	// Expand role memberships.
 	memberOf, err := p.MemberOfWithAdminOption(ctx, user)
 	if err != nil {
-		return err
+		return false, err
 	}
 
 	// Check is 'user' is a member of role 'admin'.
 	if _, ok := memberOf[sqlbase.AdminRole]; ok {
-		return nil
+		return true, nil
 	}
 
-	return pgerror.Newf(pgcode.InsufficientPrivilege,
-		"only superusers are allowed to %s", action)
+	return false, nil
+}
+
+// RequireSuperUser implements the AuthorizationAccessor interface.
+func (p *planner) RequireSuperUser(ctx context.Context, action string) error {
+	ok, err := p.IsSuperUser(ctx, action)
+
+	if err != nil {
+		return err
+	}
+	if !ok {
+		//raise error if user is not a super-user
+		return pgerror.Newf(pgcode.InsufficientPrivilege,
+			"only superusers are allowed to %s", action)
+	}
+	return nil
 }
 
 // MemberOfWithAdminOption looks up all the roles 'member' belongs to (direct and indirect) and

pkg/sql/opt/cat/catalog.go

@@ -119,6 +119,10 @@ type Catalog interface {
 	// the given catalog object. If not, then CheckAnyPrivilege returns an error.
 	CheckAnyPrivilege(ctx context.Context, o Object) error
 
+	// IsSuperUser checks that the current user has admin privileges. If yes,
+	// returns true. Returns an error if query on the `system.users` table failed
+	IsSuperUser(ctx context.Context, action string) (bool, error)
+
 	// RequireSuperUser checks that the current user has admin privileges. If not,
 	// returns an error.
 	RequireSuperUser(ctx context.Context, action string) error

pkg/sql/opt/testutils/testcat/test_catalog.go

@@ -193,6 +193,11 @@ func (tc *Catalog) CheckAnyPrivilege(ctx context.Context, o cat.Object) error {
 	return nil
 }
 
+// IsSuperUser is part of the cat.Catalog interface.
+func (tc *Catalog) IsSuperUser(ctx context.Context, action string) (bool, error) {
+	return true, nil
+}
+
 // RequireSuperUser is part of the cat.Catalog interface.
 func (tc *Catalog) RequireSuperUser(ctx context.Context, action string) error {
 	return nil

pkg/sql/opt_catalog.go

@@ -234,6 +234,11 @@ func (oc *optCatalog) CheckAnyPrivilege(ctx context.Context, o cat.Object) error
 	}
 }
 
+// IsSuperUser is part of the cat.Catalog interface.
+func (oc *optCatalog) IsSuperUser(ctx context.Context, action string) (bool, error) {
+	return oc.planner.IsSuperUser(ctx, action)
+}
+
 // RequireSuperUser is part of the cat.Catalog interface.
 func (oc *optCatalog) RequireSuperUser(ctx context.Context, action string) error {
 	return oc.planner.RequireSuperUser(ctx, action)