merged the fix for bug 30468 to 5.1-opt

mysql-test/r/grant2.result

@@ -422,4 +422,22 @@ revoke all privileges, grant option from mysqltest_1@localhost;
 revoke all privileges, grant option from mysqltest_2@localhost;
 drop user mysqltest_1@localhost;
 drop user mysqltest_2@localhost;
+CREATE DATABASE db1;
+USE db1;
+CREATE TABLE t1 (a INT, b INT);
+INSERT INTO t1 VALUES (1,1),(2,2);
+CREATE TABLE t2 (b INT, c INT);
+INSERT INTO t2 VALUES (1,100),(2,200);
+GRANT SELECT ON t1 TO mysqltest1@localhost;
+GRANT SELECT (b) ON t2 TO mysqltest1@localhost;
+USE db1;
+SELECT c FROM t2;
+ERROR 42000: SELECT command denied to user 'mysqltest1'@'localhost' for column 'c' in table 't2'
+SELECT * FROM t2;
+ERROR 42000: SELECT command denied to user 'mysqltest1'@'localhost' for column 'c' in table 't2'
+SELECT * FROM t1 JOIN t2 USING (b);
+ERROR 42000: SELECT command denied to user 'mysqltest1'@'localhost' for column 'c' in table 't2'
+DROP TABLE db1.t1, db1.t2;
+DROP USER mysqltest1@localhost;
+DROP DATABASE db1;
 End of 5.0 tests

mysql-test/t/grant2.test

@@ -585,5 +585,37 @@ drop user mysqltest_1@localhost;
 drop user mysqltest_2@localhost;
 
 
+#
+# Bug #30468: column level privileges not respected when joining tables
+#
+CREATE DATABASE db1;
+
+USE db1;
+CREATE TABLE t1 (a INT, b INT);
+INSERT INTO t1 VALUES (1,1),(2,2);
+
+CREATE TABLE t2 (b INT, c INT);
+INSERT INTO t2 VALUES (1,100),(2,200);
+
+GRANT SELECT ON t1 TO mysqltest1@localhost; 
+GRANT SELECT (b) ON t2 TO mysqltest1@localhost;
+
+connect (conn1,localhost,mysqltest1,,);
+connection conn1;
+USE db1;
+--error ER_COLUMNACCESS_DENIED_ERROR
+SELECT c FROM t2;
+--error ER_COLUMNACCESS_DENIED_ERROR
+SELECT * FROM t2;
+--error ER_COLUMNACCESS_DENIED_ERROR
+SELECT * FROM t1 JOIN t2 USING (b);
+
+connection default;
+disconnect conn1;
+DROP TABLE db1.t1, db1.t2;
+DROP USER mysqltest1@localhost;
+DROP DATABASE db1;
+
+
 --echo End of 5.0 tests
 

sql/sql_acl.cc

@@ -3991,22 +3991,48 @@ bool check_column_grant_in_table_ref(THD *thd, TABLE_LIST * table_ref,
 }
 
 
-bool check_grant_all_columns(THD *thd, ulong want_access, GRANT_INFO *grant,
-                             const char* db_name, const char *table_name,
-                             Field_iterator *fields)
+/** 
+  @brief check if a query can access a set of columns
+
+  @param  thd  the current thread
+  @param  want_access_arg  the privileges requested
+  @param  fields an iterator over the fields of a table reference.
+  @return Operation status
+    @retval 0 Success
+    @retval 1 Falure
+  @details This function walks over the columns of a table reference 
+   The columns may originate from different tables, depending on the kind of
+   table reference, e.g. join.
+   For each table it will retrieve the grant information and will use it
+   to check the required access privileges for the fields requested from it.
+*/    
+bool check_grant_all_columns(THD *thd, ulong want_access_arg, 
+                             Field_iterator_table_ref *fields)
 {
   Security_context *sctx= thd->security_ctx;
-  GRANT_TABLE *grant_table;
-  GRANT_COLUMN *grant_column;
+  ulong want_access= want_access_arg;
+  const char *table_name= NULL;
 
-  want_access &= ~grant->privilege;
-  if (!want_access)
-    return 0;				// Already checked
+  const char* db_name; 
+  GRANT_INFO *grant;
+  GRANT_TABLE *grant_table;
 
   rw_rdlock(&LOCK_grant);
 
-  /* reload table if someone has modified any grants */
+  for (; !fields->end_of_fields(); fields->next())
+  {
+    const char *field_name= fields->name();
 
+    if (table_name != fields->table_name())
+    {
+      table_name= fields->table_name();
+      db_name= fields->db_name();
+      grant= fields->grant();
+      /* get a fresh one for each table */
+      want_access= want_access_arg & ~grant->privilege;
+      if (want_access)
+      {
+        /* reload table if someone has modified any grants */
         if (grant->version != grant_version)
         {
           grant->grant_table=
@@ -4015,23 +4041,26 @@ bool check_grant_all_columns(THD *thd, ulong want_access, GRANT_INFO *grant,
                               table_name, 0);	/* purecov: inspected */
           grant->version= grant_version;	/* purecov: inspected */
         }
-  /* The following should always be true */
-  if (!(grant_table= grant->grant_table))
-    goto err;					/* purecov: inspected */
 
-  for (; !fields->end_of_fields(); fields->next())
+        DBUG_ASSERT ((grant_table= grant->grant_table) != NULL);
+      }
+    }
+
+    if (want_access)
     {
-    const char *field_name= fields->name();
-    grant_column= column_hash_search(grant_table, field_name,
+      GRANT_COLUMN *grant_column= 
+        column_hash_search(grant_table, field_name,
                            (uint) strlen(field_name));
       if (!grant_column || (~grant_column->rights & want_access))
         goto err;
     }
+  }
   rw_unlock(&LOCK_grant);
   return 0;
 
 err:
   rw_unlock(&LOCK_grant);
+
   char command[128];
   get_privilege_desc(command, sizeof(command), want_access);
   my_error(ER_COLUMNACCESS_DENIED_ERROR, MYF(0),

sql/sql_acl.h

@@ -244,9 +244,8 @@ bool check_grant_column (THD *thd, GRANT_INFO *grant,
 			 const char *name, uint length, Security_context *sctx);
 bool check_column_grant_in_table_ref(THD *thd, TABLE_LIST * table_ref,
                                      const char *name, uint length);
-bool check_grant_all_columns(THD *thd, ulong want_access, GRANT_INFO *grant,
-                             const char* db_name, const char *table_name,
-                             Field_iterator *fields);
+bool check_grant_all_columns(THD *thd, ulong want_access, 
+                             Field_iterator_table_ref *fields);
 bool check_grant_routine(THD *thd, ulong want_access,
 			 TABLE_LIST *procs, bool is_proc, bool no_error);
 bool check_grant_db(THD *thd,const char *db);

sql/sql_base.cc

@@ -6553,10 +6553,7 @@ insert_fields(THD *thd, Name_resolution_context *context, const char *db_name,
         !any_privileges)
     {
       field_iterator.set(tables);
-      if (check_grant_all_columns(thd, SELECT_ACL, field_iterator.grant(),
-                                  field_iterator.db_name(),
-                                  field_iterator.table_name(),
-                                  &field_iterator))
+      if (check_grant_all_columns(thd, SELECT_ACL, &field_iterator))
         DBUG_RETURN(TRUE);
     }
 #endif

sql/sql_insert.cc

@@ -189,11 +189,9 @@ static int check_insert_fields(THD *thd, TABLE_LIST *table_list,
       return -1;
     }
 #ifndef NO_EMBEDDED_ACCESS_CHECKS
-    Field_iterator_table field_it;
-    field_it.set_table(table);
-    if (check_grant_all_columns(thd, INSERT_ACL, &table->grant,
-                                table->s->db.str, table->s->table_name.str,
-                                &field_it))
+    Field_iterator_table_ref field_it;
+    field_it.set(table_list);
+    if (check_grant_all_columns(thd, INSERT_ACL, &field_it))
       return -1;
 #endif
     clear_timestamp_auto_bits(table->timestamp_field_type,