Added CREATE ROLE support as well as DROP ROLE support.

mysql-test/r/acl_roles_create_and_drop_role.result

+use mysql;
+create role test_role1;
+create role test_role2, test_role3;
+select user, host, is_role from user where user like 'test';
+user	host	is_role
+drop role test_role1;
+drop role test_role2, test_role3;
+create role test_role1;
+create role test_role1;
+ERROR HY000: Operation CREATE ROLE failed for 'test_role1'
+create role test_role1, test_role2;
+ERROR HY000: Operation CREATE ROLE failed for 'test_role1'
+select user, host, is_role from user where user like 'test';
+user	host	is_role
+drop role test_role1;
+drop role test_role1;
+ERROR HY000: Operation DROP ROLE failed for 'test_role1'
+drop role test_role1, test_role2;
+ERROR HY000: Operation DROP ROLE failed for 'test_role1'
+select user, host, is_role from user where user like 'test';
+user	host	is_role

mysql-test/t/acl_roles_create_and_drop_role.test

+connect (mysql, localhost, root,,);
+use mysql;
+
+create role test_role1;
+create role test_role2, test_role3;
+
+--sorted_result
+select user, host, is_role from user where user like 'test';
+
+drop role test_role1;
+drop role test_role2, test_role3;
+
+
+create role test_role1;
+--error ER_CANNOT_USER
+create role test_role1;
+--error ER_CANNOT_USER
+create role test_role1, test_role2;
+
+--sorted_result
+select user, host, is_role from user where user like 'test';
+
+drop role test_role1;
+--error ER_CANNOT_USER
+drop role test_role1;
+--error ER_CANNOT_USER
+drop role test_role1, test_role2;
+
+--sorted_result
+select user, host, is_role from user where user like 'test';
+disconnect mysql;

sql/share/errmsg-utf8.txt

@@ -6569,3 +6569,8 @@ ER_INVALID_ROLE
 ER_INVALID_CURRENT_USER
 	eng "The current user is invalid."
 	rum "Utilizatorul curent este invalid."
+ER_INVALID_ROLE_COMMAND
+	eng "Unable to execute role related command. The user table is in invalid format."
+	rum "Comanda asupra rolurilor nu poate fi executate. Tabelul "user" este in format invalid."
+ER_ROLE_AS_USER
+	eng "The role '%s' is marked as a user '%s'@''

sql/sql_acl.cc

@@ -290,6 +290,7 @@ class ACL_ROLE :public ACL_USER_BASE
   DYNAMIC_ARRAY parent_grantee; // array of backlinks to elements granted
 
   ACL_ROLE(ACL_USER * user, MEM_ROOT *mem);
+  ACL_ROLE(const char * rolename, ulong privileges, MEM_ROOT *mem);
 
 };
 
@@ -731,7 +732,8 @@ ACL_ROLE::ACL_ROLE(ACL_USER *user, MEM_ROOT *root)
 {
 
   access= user->access;
-  sort= user->sort;
+  /* set initial role access the same as the table row privileges */
+  initial_role_access= user->access;
   this->user.str= safe_strdup_root(root, user->user.str);
   this->user.length= user->user.length;
   bzero(&role_grants, sizeof(role_grants));
@@ -739,6 +741,18 @@ ACL_ROLE::ACL_ROLE(ACL_USER *user, MEM_ROOT *root)
   flags= IS_ROLE;
 }
 
+ACL_ROLE::ACL_ROLE(const char * rolename, ulong privileges, MEM_ROOT *root) :
+  initial_role_access(privileges)
+{
+  this->access= initial_role_access;
+  this->user.str= safe_strdup_root(root, rolename);
+  this->user.length= strlen(rolename);
+  bzero(&role_grants, sizeof(role_grants));
+  bzero(&parent_grantee, sizeof(parent_grantee));
+  flags= IS_ROLE;
+}
+
+
 static
 void
 free_acl_user(ACL_USER *user)
@@ -1218,8 +1232,6 @@ static my_bool acl_load(THD *thd, TABLE_LIST *tables)
         entry->role_grants = user.role_grants;
         (void) my_init_dynamic_array(&entry->parent_grantee,
                                      sizeof(ACL_USER_BASE *), 50, 100, MYF(0));
-        /* set initial role access the same as the table row privileges */
-        entry->initial_role_access= entry->access;
         my_hash_insert(&acl_roles, (uchar *)entry);
 
         continue;
@@ -1994,6 +2006,15 @@ static void acl_update_user(const char *user, const char *host,
   }
 }
 
+static void acl_insert_role(const char *rolename, ulong privileges)
+{
+  ACL_ROLE *entry;
+
+  mysql_mutex_assert_owner(&acl_cache->lock);
+  entry= new (&mem) ACL_ROLE(rolename, privileges, &mem);
+
+  my_hash_insert(&acl_roles, (uchar *)entry);
+}
 
 static void acl_insert_user(const char *user, const char *host,
 			    const char *password, uint password_len,
@@ -2174,7 +2195,6 @@ ulong acl_get(const char *host, const char *ip,
           db_access=acl_db->access;
           if (acl_db->host.hostname)
             goto exit;                          // Fully specified. Take it
-          /* XXX is this an alright way to bypass the host table for roles? */
           if ((!host || !host[0]) && !acl_db->host.hostname && find_acl_role(user))
             goto exit;
           break; /* purecov: tested */
@@ -3081,7 +3101,8 @@ static bool test_if_create_new_users(THD *thd)
 
 static int replace_user_table(THD *thd, TABLE *table, LEX_USER &combo,
 			      ulong rights, bool revoke_grant,
-			      bool can_create_user, bool no_auto_create)
+                              bool can_create_user, bool no_auto_create,
+                              bool handle_as_role)
 {
   int error = -1;
   bool old_row_exists=0;
@@ -3104,6 +3125,13 @@ static int replace_user_table(THD *thd, TABLE *table, LEX_USER &combo,
   else
     combo.password= empty_lex_str;
 
+  /* if the user table is not up to date, we can't handle role updates */
+  if (table->s->fields <= 42 && handle_as_role)
+  {
+    my_error(ER_INVALID_ROLE_COMMAND, MYF(0));
+    DBUG_RETURN(-1);
+  }
+
    table->use_all_columns();
   table->field[0]->store(combo.host.str,combo.host.length,
                          system_charset_info);
@@ -3262,6 +3290,17 @@ static int replace_user_table(THD *thd, TABLE *table, LEX_USER &combo,
         table->field[next_field + 1]->reset();
       }
     }
+
+    /* table format checked earlier */
+    if (handle_as_role)
+    {
+      if (old_row_exists && !check_is_role(table))
+      {
+        my_error(ER_ROLE_AS_USER, MYF(0), combo.user.str, combo.user.str);
+        goto end;
+      }
+      table->field[ROLE_ASSIGN_COLUMN_IDX]->store("Y", 1, system_charset_info);
+    }
   }
 
   if (old_row_exists)
@@ -3300,6 +3339,10 @@ static int replace_user_table(THD *thd, TABLE *table, LEX_USER &combo,
   {
     acl_cache->clear(1);			// Clear privilege cache
     if (old_row_exists)
+    {
+      if (handle_as_role)
+        acl_update_role(combo.user.str, rights);
+      else
         acl_update_user(combo.user.str, combo.host.str,
                         combo.password.str, combo.password.length,
                         lex->ssl_type,
@@ -3310,6 +3353,11 @@ static int replace_user_table(THD *thd, TABLE *table, LEX_USER &combo,
                         rights,
                         &combo.plugin,
                         &combo.auth);
+    }
+    else
+    {
+      if (handle_as_role)
+        acl_insert_role(combo.user.str, rights);
       else
         acl_insert_user(combo.user.str, combo.host.str,
                         combo.password.str, combo.password.length,
@@ -3322,6 +3370,7 @@ static int replace_user_table(THD *thd, TABLE *table, LEX_USER &combo,
                         &combo.plugin,
                         &combo.auth);
     }
+  }
   DBUG_RETURN(error);
 }
 
@@ -4537,7 +4586,7 @@ int mysql_table_grant(THD *thd, TABLE_LIST *table_list,
     error=replace_user_table(thd, tables[0].table, *Str,
 			     0, revoke_grant, create_new_users,
                              test(thd->variables.sql_mode &
-                                  MODE_NO_AUTO_CREATE_USER));
+                                  MODE_NO_AUTO_CREATE_USER), 0);
     if (error)
     {
       result= TRUE;				// Remember error
@@ -4743,7 +4792,7 @@ bool mysql_routine_grant(THD *thd, TABLE_LIST *table_list, bool is_proc,
     error=replace_user_table(thd, tables[0].table, *Str,
 			     0, revoke_grant, create_new_users,
                              test(thd->variables.sql_mode &
-                                  MODE_NO_AUTO_CREATE_USER));
+                                  MODE_NO_AUTO_CREATE_USER), 0);
     if (error)
     {
       result= TRUE;				// Remember error
@@ -4900,7 +4949,7 @@ bool mysql_grant(THD *thd, const char *db, List <LEX_USER> &list,
     if (replace_user_table(thd, tables[0].table, *Str,
                            (!db ? rights : 0), revoke_grant, create_new_users,
                            test(thd->variables.sql_mode &
-                                MODE_NO_AUTO_CREATE_USER)))
+                                MODE_NO_AUTO_CREATE_USER), 0))
       result= -1;
     else if (db)
     {
@@ -6629,7 +6678,7 @@ void get_mqh(const char *user, const char *host, USER_CONN *uc)
     1 db
     2 tables_priv
     3 columns_priv
-    4 columns_priv
+    4 procs_priv
     5 proxies_priv
     6 roles_mapping
 
@@ -7552,6 +7601,15 @@ static void append_user(String *str, LEX_USER *user)
   str->append('\'');
 }
 
+static void append_role(String *str, LEX_USER *user)
+{
+  if (str->length())
+    str->append(',');
+  str->append('\'');
+  str->append(user->user.str);
+  str->append('\'');
+}
+
 
 /*
   Create a list of users.
@@ -7603,7 +7661,7 @@ bool mysql_create_user(THD *thd, List <LEX_USER> &list)
     }
 
     some_users_created= TRUE;
-    if (replace_user_table(thd, tables[0].table, *user_name, 0, 0, 1, 0))
+    if (replace_user_table(thd, tables[0].table, *user_name, 0, 0, 1, 0, 0))
     {
       append_user(&wrong_users, user_name);
       result= TRUE;
@@ -7622,6 +7680,69 @@ bool mysql_create_user(THD *thd, List <LEX_USER> &list)
   DBUG_RETURN(result);
 }
 
+/*
+  Create a list of roles.
+
+  SYNOPSIS
+    mysql_create_role()
+    thd                         The current thread.
+    list                        The users to create.
+
+  RETURN
+    FALSE       OK.
+    TRUE        Error.
+*/
+
+bool mysql_create_role(THD *thd, List <LEX_USER> &list)
+{
+  int result;
+  String wrong_users;
+  LEX_USER *role_name;
+  List_iterator <LEX_USER> role_list(list);
+  TABLE_LIST tables[GRANT_TABLES];
+  bool some_users_created= FALSE;
+  DBUG_ENTER("mysql_create_role");
+
+  if ((result= open_grant_tables(thd, tables)))
+    DBUG_RETURN(result != 1);
+
+  mysql_rwlock_wrlock(&LOCK_grant);
+  mysql_mutex_lock(&acl_cache->lock);
+
+  while ((role_name= role_list++))
+  {
+    role_name->host.str= (char *)"";
+    role_name->host.length= 0;
+    /*
+      Search all in-memory structures and grant tables
+      for a mention of the new user name.
+    */
+    if (handle_grant_data(tables, 0, role_name, NULL))
+    {
+      append_role(&wrong_users, role_name);
+      result= TRUE;
+      continue;
+    }
+    some_users_created= TRUE;
+    if (replace_user_table(thd, tables[0].table, *role_name, 0, 0, 1, 0, 1))
+    {
+      append_role(&wrong_users, role_name);
+      result= TRUE;
+    }
+  }
+
+  mysql_mutex_unlock(&acl_cache->lock);
+
+  if (result)
+    my_error(ER_CANNOT_USER, MYF(0), "CREATE ROLE", wrong_users.c_ptr_safe());
+
+  if (some_users_created)
+    result |= write_bin_log(thd, FALSE, thd->query(), thd->query_length());
+
+  mysql_rwlock_unlock(&LOCK_grant);
+DBUG_RETURN(result);
+}
+
 
 /*
   Drop a list of users and all their privileges.
@@ -7691,6 +7812,68 @@ bool mysql_drop_user(THD *thd, List <LEX_USER> &list)
   DBUG_RETURN(result);
 }
 
+/*
+  Drop a list of roles and all their privileges.
+
+  SYNOPSIS
+    mysql_drop_role()
+    thd                         The current thread.
+    list                        The roles to drop.
+
+  RETURN
+    FALSE       OK.
+    TRUE        Error.
+*/
+bool mysql_drop_role(THD *thd, List <LEX_USER> &list)
+{
+  int result;
+  String wrong_users;
+  LEX_USER *role_name;
+  List_iterator <LEX_USER> user_list(list);
+  TABLE_LIST tables[GRANT_TABLES];
+  bool some_users_deleted= FALSE;
+  ulonglong old_sql_mode= thd->variables.sql_mode;
+  DBUG_ENTER("mysql_drop_role");
+
+  /* DROP USER may be skipped on replication client. */
+  if ((result= open_grant_tables(thd, tables)))
+    DBUG_RETURN(result != 1);
+
+  thd->variables.sql_mode&= ~MODE_PAD_CHAR_TO_FULL_LENGTH;
+
+  mysql_rwlock_wrlock(&LOCK_grant);
+  mysql_mutex_lock(&acl_cache->lock);
+
+  while ((role_name= user_list++))
+  {
+    role_name->host.str= (char *)"";
+    role_name->host.length= 0;
+
+    if (handle_grant_data(tables, 1, role_name, NULL) <= 0)
+    {
+      append_role(&wrong_users, role_name);
+      result= TRUE;
+      continue;
+    }
+    some_users_deleted= TRUE;
+  }
+
+  /* Rebuild every user's role_grants because the acl_role has been modified
+     and some grants might now be invalid */
+  rebuild_role_grants();
+
+  mysql_mutex_unlock(&acl_cache->lock);
+
+  if (result)
+    my_error(ER_CANNOT_USER, MYF(0), "DROP ROLE", wrong_users.c_ptr_safe());
+
+  if (some_users_deleted)
+    result |= write_bin_log(thd, FALSE, thd->query(), thd->query_length());
+
+  mysql_rwlock_unlock(&LOCK_grant);
+  thd->variables.sql_mode= old_sql_mode;
+  DBUG_RETURN(result);
+}
 
 /*
   Rename a user.
@@ -7819,7 +8002,7 @@ bool mysql_revoke_all(THD *thd,  List <LEX_USER> &list)
     }
 
     if (replace_user_table(thd, tables[0].table,
-			   *lex_user, ~(ulong)0, 1, 0, 0))
+                           *lex_user, ~(ulong)0, 1, 0, 0, 0))
     {
       result= -1;
       continue;

sql/sql_acl.h

@@ -234,6 +234,8 @@ bool mysql_create_user(THD *thd, List <LEX_USER> &list);
 bool mysql_drop_user(THD *thd, List <LEX_USER> &list);
 bool mysql_rename_user(THD *thd, List <LEX_USER> &list);
 bool mysql_revoke_all(THD *thd, List <LEX_USER> &list);
+bool mysql_create_role(THD *thd, List <LEX_USER> &list);
+bool mysql_drop_role(THD *thd, List <LEX_USER> &list);
 void fill_effective_table_privileges(THD *thd, GRANT_INFO *grant,
                                      const char *db, const char *table);
 bool sp_revoke_privileges(THD *thd, const char *sp_db, const char *sp_name,

sql/sql_parse.cc

@@ -3759,13 +3759,21 @@ case SQLCOM_PREPARE:
   }
   case SQLCOM_CREATE_ROLE:
   {
-    /* TODO */
+    if (check_access(thd, INSERT_ACL, "mysql", NULL, NULL, 1, 1) &&
+        check_global_access(thd,CREATE_USER_ACL))
+      break;
+    /* Conditionally writes to binlog */
+    if (!(res= mysql_create_role(thd, lex->users_list)))
       my_ok(thd);
     break;
   }
   case SQLCOM_DROP_ROLE:
   {
-    /* TODO */
+    if (check_access(thd, DELETE_ACL, "mysql", NULL, NULL, 1, 1) &&
+        check_global_access(thd,CREATE_USER_ACL))
+      break;
+    /* Conditionally writes to binlog */
+    if (!(res= mysql_drop_role(thd, lex->users_list)))
       my_ok(thd);
     break;
   }

sql/sql_yacc.yy

@@ -1459,7 +1459,6 @@ bool my_yyoverflow(short **a, YYSTYPE **b, ulong *yystacksize);
         NCHAR_STRING opt_component key_cache_name
         sp_opt_label BIN_NUM label_ident TEXT_STRING_filesystem ident_or_empty
         opt_constraint constraint opt_ident opt_if_not_exists_ident
-        grant_role
 
 %type <lex_str_ptr>
         opt_table_alias
@@ -1570,7 +1569,7 @@ bool my_yyoverflow(short **a, YYSTYPE **b, ulong *yystacksize);
 
 %type <symbol> keyword keyword_sp
 
-%type <lex_user> user grant_user
+%type <lex_user> user grant_user grant_role
 
 %type <charset>
         opt_collate
@@ -14251,7 +14250,6 @@ revoke_command:
             LEX *lex= Lex;
             lex->sql_command= SQLCOM_REVOKE_ROLE;
             lex->type= 0;
-            printf("The rolename to be revoked is: %s\n", $1.str);
           }
 
         ;
@@ -14305,20 +14303,40 @@ grant_command:
             LEX *lex= Lex;
             lex->sql_command= SQLCOM_GRANT_ROLE;
             lex->type= 0;
-            printf("The rolename to be granted is: %s\n", $1.str);
           }
 
         ;
 
 role_list:
           grant_role
-          {}
+          {
+            if (Lex->users_list.push_back($1))
+              MYSQL_YYABORT;
+          }
         | role_list ',' grant_role
-          {}
+          {
+            if (Lex->users_list.push_back($3))
+              MYSQL_YYABORT;
+          }
+        ;
 
 grant_role:
-          IDENT_sys             {$$=$1;}
-        | TEXT_STRING_sys       {$$=$1;}
+          ident_or_text
+          {
+            if (!($$=(LEX_USER*) thd->alloc(sizeof(st_lex_user))))
+              MYSQL_YYABORT;
+            $$->user = $1;
+            $$->host.str= (char *)HOST_NOT_SPECIFIED;
+            $$->host.length= 1;
+            $$->password= null_lex_str; 
+            $$->plugin= empty_lex_str;
+            $$->auth= empty_lex_str;
+
+            if (check_string_char_length(&$$->user, ER(ER_USERNAME),
+                                         username_char_length,
+                                         system_charset_info, 0))
+              MYSQL_YYABORT;
+          }
         ;
 
 opt_table: