Re: New patch for Column-level privileges - Mailing list pgsql-hackers

>>> Currently,
>>> column-level privileges are not honored when JOINs are involved (you
>>> must have the necessary table-level privileges, as you do today). It
>>> would really be great to have that working and mainly involves
>>> modifying the rewriter to add on to the appropriate range table column
>>> list entries the columns which are used in the joins and output from
>>> joins.
>>
>> Understood. Do you plan to work on that? Or do you think the patch
>> should go into 8.4 even without support for JOINs?
>
> I tried to check the latest Column-level privileges patch.
>
> This trouble related to JOINs is come from inappropriate handling
> of rte->cols_sel list, in my opinion.
> The list is constructed at scanRTEForColumn() and expandRelation().
> However, scanRTEForColumn() appends an appeared attribute number
> even if given RangeTblEntry is RTE_JOIN, and expandRelation() appends
> all the attribute number contained within the given relation.
>
> I think your design is basically fine, so the issue is minor one.
> But it is necessary to pick up carefully what columns are really
> accessed.
>
> Here is one proposition.
> Is it possible to implement a walker function to pick up appeared
> columns and to chain them on rte->cols_sel/cols_mod?
> In this idea, columns in Query->targetList should be chained on
> rte->cols_mod, and others should be chained on rte->cols_sel.
>
> Here is an example to pick up all appeared relation:
> http://code.google.com/p/sepgsql/source/browse/trunk/sepgsql/src/backend/security/rowacl/rowacl.c#30
>
>
> If you don't have enough availability, I'll be able to do it within
> a few days.

The attached patch is a proof of the concept.
It walks on a given query tree to append accessed columns on
rte->cols_sel and rte->cols_mod.
When aliasvar of JOIN'ed relation is accesses, its source is
appended on the list.

This patch can be applied on the latest CVS HEAD with Stephen's
colprivs_wip.2009010201.diff.gz.

Any comment?

I strongly want the Column-level privileges to be get merged
as soon as possible, so I don't spare any possible assist
for his works.

Thanks,

---- example of the patched Column-level privileges ----
postgres=# CREATE TABLE t1 (a int, b text, c bool);
CREATE TABLE
postgres=# CREATE TABLE t2 (x int, y text, z bool);
CREATE TABLE
postgres=# GRANT SELECT(a,b) ON t1 TO ymj;
GRANT
postgres=# GRANT UPDATE(a,b) ON t1 TO ymj;
GRANT
postgres=# GRANT SELECT(x,y) ON t2 TO ymj;
GRANT
postgres=# INSERT INTO t1 VALUES (1, 'aaa', true), (2, 'bbb', null), (3, 'ccc', false);
INSERT 0 3
postgres=# INSERT INTO t2 VALUES (1, 'xxx', false), (2, 'yyy', null), (3, 'zzz', true);
INSERT 0 3
postgres=# \c - ymj
psql (8.4devel)
You are now connected to database "postgres" as user "ymj".
postgres=> SELECT * FROM t1;
NOTICE:  pg_attribute_aclmask: t1.a required: 0002 allowed: 0002
NOTICE:  pg_attribute_aclmask: t1.b required: 0002 allowed: 0002
NOTICE:  pg_attribute_aclmask: t1.c required: 0002 allowed: 0000
ERROR:  permission denied for relation t1
postgres=> SELECT a,b FROM t1;
NOTICE:  pg_attribute_aclmask: t1.a required: 0002 allowed: 0002
NOTICE:  pg_attribute_aclmask: t1.b required: 0002 allowed: 0002
  a |  b
---+-----
  1 | aaa
  2 | bbb
  3 | ccc
(3 rows)

postgres=> SELECT x,y FROM t2;
NOTICE:  pg_attribute_aclmask: t2.x required: 0002 allowed: 0002
NOTICE:  pg_attribute_aclmask: t2.y required: 0002 allowed: 0002
  x |  y
---+-----
  1 | xxx
  2 | yyy
  3 | zzz
(3 rows)

postgres=> SELECT b,y FROM t1 JOIN t2 ON a = x;
NOTICE:  pg_attribute_aclmask: t1.b required: 0002 allowed: 0002
NOTICE:  pg_attribute_aclmask: t1.a required: 0002 allowed: 0002
NOTICE:  pg_attribute_aclmask: t2.y required: 0002 allowed: 0002
NOTICE:  pg_attribute_aclmask: t2.x required: 0002 allowed: 0002
   b  |  y
-----+-----
  aaa | xxx
  bbb | yyy
  ccc | zzz
(3 rows)

postgres=> SELECT b,z FROM t1 JOIN t2 ON a = x;
NOTICE:  pg_attribute_aclmask: t1.b required: 0002 allowed: 0002
NOTICE:  pg_attribute_aclmask: t1.a required: 0002 allowed: 0002
NOTICE:  pg_attribute_aclmask: t2.z required: 0002 allowed: 0000
ERROR:  permission denied for relation t2
postgres=> UPDATE t1 SET a = 1;
NOTICE:  pg_attribute_aclmask: t1.a required: 0004 allowed: 0004
UPDATE 3
postgres=> UPDATE t1 SET c = true;
NOTICE:  pg_attribute_aclmask: t1.c required: 0004 allowed: 0000
ERROR:  permission denied for relation t1
postgres=>

--
OSS Platform Development Division, NEC
KaiGai Kohei <kaigai@ak.jp.nec.com>
Index: src/backend/parser/analyze.c
===================================================================
*** src/backend/parser/analyze.c    (revision 1)
--- src/backend/parser/analyze.c    (working copy)
***************
*** 27,32 ****
--- 27,33 ----
  #include "catalog/pg_type.h"
  #include "nodes/makefuncs.h"
  #include "nodes/nodeFuncs.h"
+ #include "optimizer/tlist.h"
  #include "optimizer/var.h"
  #include "parser/analyze.h"
  #include "parser/parse_agg.h"
***************
*** 267,272 ****
--- 268,338 ----
  }

  /*
+  * applyColumnPrivs()
+  *     construct rte->cols_sel and rte->cols_mod for Column-level
+  *     privileges.
+  */
+ static bool
+ applyColumnPrivsWalker(Node *node, ParseState *pstate)
+ {
+     if (!node)
+         return false;
+
+     if (IsA(node, Var))
+     {
+         RangeTblEntry *rte;
+         Var *v = (Var *) node;
+         int lv;
+
+         for (lv = v->varlevelsup; lv > 0; lv--)
+         {
+             Assert(pstate->parentParseState != NULL);
+             pstate = pstate->parentParseState;
+         }
+
+         rte = rt_fetch(v->varno, pstate->p_rtable);
+         Assert(IsA(rte, RangeTblEntry));
+
+         if (rte->rtekind == RTE_RELATION)
+         {
+             rte->cols_sel = lappend_int(rte->cols_sel, v->varattno);
+         }
+         else if (rte->rtekind == RTE_JOIN)
+         {
+             node = list_nth(rte->joinaliasvars, v->varattno - 1);
+
+             applyColumnPrivsWalker(node, pstate);
+         }
+     }
+     else if (IsA(node, SortGroupClause))
+         return false;
+
+     return expression_tree_walker(node, applyColumnPrivsWalker, (void *) pstate);
+ }
+
+ static void
+ applyColumnPrivs(Query *query, ParseState *pstate)
+ {
+     ListCell *l;
+
+     if (query->commandType != CMD_SELECT)
+     {
+         RangeTblEntry *rte
+             = rt_fetch(query->resultRelation, query->rtable);
+
+         foreach (l, query->targetList)
+         {
+             TargetEntry *tle = lfirst(l);
+
+             Assert(IsA(tle, TargetEntry));
+             rte->cols_mod = lappend_int(rte->cols_mod, tle->resno);
+         }
+     }
+     query_tree_walker(query, applyColumnPrivsWalker, (void *) pstate,
+                       QTW_IGNORE_JOINALIASES);
+ }
+
+ /*
   * transformDeleteStmt -
   *      transforms a Delete Statement
   */
***************
*** 683,688 ****
--- 749,756 ----
                   parser_errposition(pstate,
                                      locate_windowfunc((Node *) qry))));

+     applyColumnPrivs(qry, pstate);
+
      return qry;
  }

***************
*** 876,881 ****
--- 944,951 ----
          transformLockingClause(pstate, qry, (LockingClause *) lfirst(l));
      }

+     applyColumnPrivs(qry, pstate);
+
      return qry;
  }

***************
*** 1716,1721 ****
--- 1786,1793 ----
      if (origTargetList != NULL)
          elog(ERROR, "UPDATE target count mismatch --- internal error");

+     applyColumnPrivs(qry, pstate);
+
      return qry;
  }

Index: src/backend/parser/parse_relation.c
===================================================================
*** src/backend/parser/parse_relation.c    (revision 2)
--- src/backend/parser/parse_relation.c    (working copy)
***************
*** 479,485 ****
              result = (Node *) make_var(pstate, rte, attnum, location);
              /* Require read access */
              rte->requiredPerms |= ACL_SELECT;
-             rte->cols_sel = lappend_int(rte->cols_sel,attnum);
          }
      }

--- 479,484 ----
***************
*** 508,514 ****
                  result = (Node *) make_var(pstate, rte, attnum, location);
                  /* Require read access */
                  rte->requiredPerms |= ACL_SELECT;
-                 rte->cols_sel = lappend_int(rte->cols_sel,attnum);
              }
          }
      }
--- 507,512 ----
***************
*** 1749,1762 ****
      expandTupleDesc(rel->rd_att, rte->eref, rtindex, sublevels_up,
                      location, include_dropped,
                      colnames, colvars);
-
-     for (attrno = 0; attrno < rel->rd_att->natts; attrno++)
-     {
-         Form_pg_attribute attr = rel->rd_att->attrs[attrno];
-
-         if (!attr->attisdropped)
-             rte->cols_sel = lappend_int(rte->cols_sel, attrno+1);
-     }
      relation_close(rel, AccessShareLock);
  }

--- 1747,1752 ----
Index: src/backend/parser/parse_target.c
===================================================================
*** src/backend/parser/parse_target.c    (revision 2)
--- src/backend/parser/parse_target.c    (working copy)
***************
*** 372,378 ****
      attrtype = attnumTypeId(rd, attrno);
      attrtypmod = rd->rd_att->attrs[attrno - 1]->atttypmod;

-     pstate->p_target_rangetblentry->cols_mod = lappend_int(pstate->p_target_rangetblentry->cols_mod, attrno);

      /*
       * If the expression is a DEFAULT placeholder, insert the attribute's
--- 372,377 ----
***************
*** 785,791 ****
              col->location = -1;
              cols = lappend(cols, col);
              *attrnos = lappend_int(*attrnos, i + 1);
-             pstate->p_target_rangetblentry->cols_mod = lappend_int(pstate->p_target_rangetblentry->cols_mod, i + 1);
          }
      }
      else
--- 784,789 ----
***************
*** 842,848 ****
              }

              *attrnos = lappend_int(*attrnos, attrno);
-             pstate->p_target_rangetblentry->cols_mod = lappend_int(pstate->p_target_rangetblentry->cols_mod, attrno);
          }
      }

--- 840,845 ----
Index: src/backend/catalog/aclchk.c
===================================================================
*** src/backend/catalog/aclchk.c    (revision 2)
--- src/backend/catalog/aclchk.c    (working copy)
***************
*** 2291,2296 ****
--- 2291,2302 ----

      pfree(acl);

+     elog(NOTICE, "%s: %s.%s required: %04x allowed: %04x",
+          __FUNCTION__,
+          NameStr(classForm->relname),
+          NameStr(((Form_pg_attribute) GETSTRUCT(attTuple))->attname),
+          mask, result);
+
      /* if we have a detoasted copy, free it */
      if (colacl && (Pointer) colacl != DatumGetPointer(colaclDatum))
          pfree(colacl);