Re: [BUG] Column-level privileges on inherited tables - Mailing list pgsql-hackers

Stephen,

The attached patch fixes the matter.
It fixes up attribute number of child relation when it is extracted.

(*) Injected elog()s are removed.

postgres=# select * from t1;
NOTICE:  markRTEForSelectPriv: ACL_SELECT on t1.a
NOTICE:  markRTEForSelectPriv: ACL_SELECT on t1.c
NOTICE:  ExecCheckRTEPerms: ACL_SELECT on t1 perms = 0000 inh = 1
NOTICE:  ExecCheckRTEPerms: selectedCols: t1.a
NOTICE:  ExecCheckRTEPerms: selectedCols: t1.c
NOTICE:  ExecCheckRTEPerms: ACL_SELECT on t1 perms = 0002 inh = 0
NOTICE:  ExecCheckRTEPerms: selectedCols: t1.a
NOTICE:  ExecCheckRTEPerms: selectedCols: t1.c
NOTICE:  ExecCheckRTEPerms: ACL_SELECT on t2 perms = 0002 inh = 0
NOTICE:  ExecCheckRTEPerms: selectedCols: t2.a
NOTICE:  ExecCheckRTEPerms: selectedCols: t2.c
 a | c
---+---
(0 rows)

postgres=# select t1 from t1;
NOTICE:  markRTEForSelectPriv: ACL_SELECT on t1.t1
NOTICE:  ExecCheckRTEPerms: ACL_SELECT on t1 perms = 0000 inh = 1
NOTICE:  ExecCheckRTEPerms: selectedCols: t1.t1
NOTICE:  ExecCheckRTEPerms: ACL_SELECT on t1 perms = 0002 inh = 0
NOTICE:  ExecCheckRTEPerms: selectedCols: t1.t1
NOTICE:  ExecCheckRTEPerms: ACL_SELECT on t2 perms = 0002 inh = 0
NOTICE:  ExecCheckRTEPerms: selectedCols: t2.a
NOTICE:  ExecCheckRTEPerms: selectedCols: t2.c
 t1
----
(0 rows)

KaiGai Kohei wrote:
> KaiGai Kohei wrote:
>> I've observed the behavior of column-level privileges and
>> required permissions with a few elog()s injected.
>>
>> I noticed rte->selectedCols is incorrect when we make a query
>> on inherited tables.
>>
>> See below:
>> -------------------------------------------------
>> postgres=# CREATE TABLE t1 (a int, b int, c int);
>> CREATE TABLE
>> postgres=# ALTER TABLE t1 DROP COLUMN b;
>> ALTER TABLE
>> postgres=# CREATE TABLE t2 (d int) inherits (t1);
>> CREATE TABLE
>> postgres=# SELECT * FROM t1;
>> NOTICE:  markRTEForSelectPriv: ACL_SELECT on t1.a
>> NOTICE:  markRTEForSelectPriv: ACL_SELECT on t1.c
>> NOTICE:  ExecCheckRTEPerms: ACL_SELECT on t1 perms = 0000 inh = 1
>> NOTICE:  ExecCheckRTEPerms: selectedCols: t1.a
>> NOTICE:  ExecCheckRTEPerms: selectedCols: t1.c
>> NOTICE:  ExecCheckRTEPerms: ACL_SELECT on t1 perms = 0002 inh = 0
>> NOTICE:  ExecCheckRTEPerms: selectedCols: t1.a
>> NOTICE:  ExecCheckRTEPerms: selectedCols: t1.c
>> NOTICE:  ExecCheckRTEPerms: ACL_SELECT on t2 perms = 0002 inh = 0
>> NOTICE:  ExecCheckRTEPerms: selectedCols: t2.a
>> NOTICE:  ExecCheckRTEPerms: selectedCols: t2.d  <--- (*)
>>  a | c
>> ---+---
>> (0 rows)
>> -------------------------------------------------
>>
>> I injected elog() at the head of ExecCheckRTEPerms() to print requiredPerms
>> and all the columns on selectedCols/modifiedCols.
>>
>> It seems to me the current implementation assumes the parant table and
>> child table have same set of attribute name/number pair, but incorrect.
>> It is necessary to lookup attribute names of "t2" when we extract
>> inherited tables.
>
> In addition, the whole-row-reference can be problematic.
>
> When we run "SELECT t1 FROM t1", the column level privilege tries to check
> all the columns within t1 and t2. But, I think it should not check on "t2.d"
> in this case, because the column is not a target of this query.
>
> In the whole-row-reference case, attno==0 on the parent table should be
> extracted into correct set of columns on the children.
>
> Thanks,

--
OSS Platform Development Division, NEC
KaiGai Kohei <kaigai@ak.jp.nec.com>
*** base/src/backend/optimizer/prep/prepunion.c    2009-02-26 11:04:20.000000000 +0900
--- sepgsql/src/backend/optimizer/prep/prepunion.c    2009-03-05 16:24:32.000000000 +0900
***************
*** 30,35 ****
--- 30,36 ----


  #include "access/heapam.h"
+ #include "access/sysattr.h"
  #include "catalog/namespace.h"
  #include "catalog/pg_type.h"
  #include "miscadmin.h"
***************
*** 49,54 ****
--- 50,56 ----
  #include "utils/lsyscache.h"
  #include "utils/rel.h"
  #include "utils/selfuncs.h"
+ #include "utils/syscache.h"


  static Plan *recurse_set_operations(Node *setOp, PlannerInfo *root,
***************
*** 1150,1155 ****
--- 1152,1253 ----
  }

  /*
+  * fixup_column_privileges
+  *   Inherited tables have same columns as its parents have,
+  *   but these columns may have different attribute numbers,
+  *   so we need to lookup attribute numbers of child relation
+  *   by its name.
+  */
+ static Bitmapset *
+ fixup_column_privileges(Oid parent, Oid child, Bitmapset *oldmap)
+ {
+     Bitmapset  *newmap = NULL;
+     char       *attname;
+     int            attno, attno_fixup;
+
+     if (!oldmap || parent == child)
+         return oldmap;    /* no need to fixup */
+
+     while ((attno = bms_first_member(oldmap)) >= 0)
+     {
+         /* remove the column number offset */
+         attno += FirstLowInvalidHeapAttributeNumber;
+
+         /*
+          * The whole-row-reference case need a special care
+          * because child relation has more columns than the
+          * parent, so it need to extract inherited columns
+          * only.
+          */
+         if (attno == InvalidAttrNumber)
+         {
+             HeapTuple            reltup;
+             HeapTuple            atttup;
+             Form_pg_class        classForm;
+             Form_pg_attribute    attForm;
+             int                    nattrs;
+
+             reltup = SearchSysCache(RELOID,
+                                     ObjectIdGetDatum(parent),
+                                     0, 0, 0);
+             if (!HeapTupleIsValid(reltup))
+                 elog(ERROR, "cache lookup failed for relation %u", parent);
+             classForm = (Form_pg_class) GETSTRUCT(reltup);
+
+             nattrs = classForm->relnatts;
+
+             ReleaseSysCache(reltup);
+
+             for (attno = 1; attno <= nattrs; attno++)
+             {
+                 atttup = SearchSysCache(ATTNUM,
+                                         ObjectIdGetDatum(parent),
+                                         Int16GetDatum(attno),
+                                         0, 0);
+                 if (!HeapTupleIsValid(atttup))
+                     continue;
+
+                 attForm = (Form_pg_attribute) GETSTRUCT(atttup);
+
+                 if (attForm->attisdropped)
+                 {
+                     ReleaseSysCache(atttup);
+                     continue;
+                 }
+
+                 attno_fixup = get_attnum(child, NameStr(attForm->attname));
+                 if (attno_fixup == InvalidAttrNumber)
+                     elog(ERROR, "cache lookup failed for attribute %s of relation %u",
+                          NameStr(attForm->attname), child);
+
+                 /* add the column number offset */
+                 attno_fixup -= FirstLowInvalidHeapAttributeNumber;
+                 newmap = bms_add_member(newmap, attno_fixup);
+
+                 ReleaseSysCache(atttup);
+             }
+             continue;
+         }
+
+         attname = get_attname(parent, attno);
+         if (!attname)
+             elog(ERROR, "cache lookup failed for attribute %d of relation %u",
+                  attno, parent);
+
+         attno_fixup = get_attnum(child, attname);
+         if (attno_fixup == InvalidAttrNumber)
+             elog(ERROR, "cache lookup failed for attribute %s of relation %u",
+                  attname, child);
+
+         /* add the column number offset */
+         attno_fixup -= FirstLowInvalidHeapAttributeNumber;
+         newmap = bms_add_member(newmap, attno_fixup);
+     }
+
+     return newmap;
+ }
+
+ /*
   * expand_inherited_rtentry
   *        Check whether a rangetable entry represents an inheritance set.
   *        If so, add entries for all the child tables to the query's
***************
*** 1277,1282 ****
--- 1375,1384 ----
           * and set inh = false.
           */
          childrte = copyObject(rte);
+         childrte->selectedCols
+             = fixup_column_privileges(rte->relid, childOID, childrte->selectedCols);
+         childrte->modifiedCols
+             = fixup_column_privileges(rte->relid, childOID, childrte->modifiedCols);
          childrte->relid = childOID;
          childrte->inh = false;
          parse->rtable = lappend(parse->rtable, childrte);