Fix has_column_privilege function corner case

According to the comments, when an invalid or dropped column oid is passed
to has_column_privilege(), the intention has always been to return NULL.
However, when the caller had table level privilege the invalid/missing
column was never discovered, because table permissions were checked first.

Fix that by introducing extended versions of pg_attribute_acl(check|mask)
and pg_class_acl(check|mask) which take a new argument, is_missing. When
is_missing is NULL, the old behavior is preserved. But when is_missing is
passed by the caller, no ERROR is thrown for dropped or missing
columns/relations, and is_missing is flipped to true. This in turn allows
has_column_privilege to check for column privileges first, providing the
desired semantics.

Not backpatched since it is a user visible behavioral change with no previous
complaints, and the fix is a bit on the invasive side.

Author: Joe Conway
Reviewed-By: Tom Lane
Reported by: Ian Barwick
Discussion: https://postgr.es/m/flat/9b5f4311-157b-4164-7fe7-077b4fe8ed84%40joeconway.com

diff --git a/src/backend/catalog/aclchk.c b/src/backend/catalog/aclchk.c
index add3d147e766cde6ee64af335cc1f6988f5c66f6..4b2afffb8faa4ca9475347774e5f46599078c5da 100644 (file)
--- a/src/backend/catalog/aclchk.c
+++ b/src/backend/catalog/aclchk.c
@@ -3705,6 +3705,20 @@ pg_aclmask(ObjectType objtype, Oid table_oid, AttrNumber attnum, Oid roleid,
 AclMode
 pg_attribute_aclmask(Oid table_oid, AttrNumber attnum, Oid roleid,
                                         AclMode mask, AclMaskHow how)
+{
+       return pg_attribute_aclmask_ext(table_oid, attnum, roleid,
+                                                                       mask, how, NULL);
+}
+
+/*
+ * Exported routine for examining a user's privileges for a column
+ *
+ * Does the bulk of the work for pg_attribute_aclmask(), and allows other
+ * callers to avoid the missing attribute ERROR when is_missing is non-NULL.
+ */
+AclMode
+pg_attribute_aclmask_ext(Oid table_oid, AttrNumber attnum, Oid roleid,
+                                                AclMode mask, AclMaskHow how, bool *is_missing)
 {
        AclMode         result;
        HeapTuple       classTuple;
@@ -3723,18 +3737,38 @@ pg_attribute_aclmask(Oid table_oid, AttrNumber attnum, Oid roleid,
                                                           ObjectIdGetDatum(table_oid),
                                                           Int16GetDatum(attnum));
        if (!HeapTupleIsValid(attTuple))
-               ereport(ERROR,
-                               (errcode(ERRCODE_UNDEFINED_COLUMN),
-                                errmsg("attribute %d of relation with OID %u does not exist",
-                                               attnum, table_oid)));
+       {
+               if (is_missing != NULL)
+               {
+                       /* return "no privileges" instead of throwing an error */
+                       *is_missing = true;
+                       return 0;
+               }
+               else
+                       ereport(ERROR,
+                                       (errcode(ERRCODE_UNDEFINED_COLUMN),
+                                        errmsg("attribute %d of relation with OID %u does not exist",
+                                                       attnum, table_oid)));
+       }
+
        attributeForm = (Form_pg_attribute) GETSTRUCT(attTuple);
 
-       /* Throw error on dropped columns, too */
+       /* Check dropped columns, too */
        if (attributeForm->attisdropped)
-               ereport(ERROR,
-                               (errcode(ERRCODE_UNDEFINED_COLUMN),
-                                errmsg("attribute %d of relation with OID %u does not exist",
-                                               attnum, table_oid)));
+       {
+               if (is_missing != NULL)
+               {
+                       /* return "no privileges" instead of throwing an error */
+                       *is_missing = true;
+                       ReleaseSysCache(attTuple);
+                       return 0;
+               }
+               else
+                       ereport(ERROR,
+                                       (errcode(ERRCODE_UNDEFINED_COLUMN),
+                                        errmsg("attribute %d of relation with OID %u does not exist",
+                                                       attnum, table_oid)));
+       }
 
        aclDatum = SysCacheGetAttr(ATTNUM, attTuple, Anum_pg_attribute_attacl,
                                                           &isNull);
@@ -3790,6 +3824,19 @@ pg_attribute_aclmask(Oid table_oid, AttrNumber attnum, Oid roleid,
 AclMode
 pg_class_aclmask(Oid table_oid, Oid roleid,
                                 AclMode mask, AclMaskHow how)
+{
+       return pg_class_aclmask_ext(table_oid, roleid, mask, how, NULL);
+}
+
+/*
+ * Exported routine for examining a user's privileges for a table
+ *
+ * Does the bulk of the work for pg_class_aclmask(), and allows other
+ * callers to avoid the missing relation ERROR when is_missing is non-NULL.
+ */
+AclMode
+pg_class_aclmask_ext(Oid table_oid, Oid roleid, AclMode mask,
+                                        AclMaskHow how, bool *is_missing)
 {
        AclMode         result;
        HeapTuple       tuple;
@@ -3804,10 +3851,20 @@ pg_class_aclmask(Oid table_oid, Oid roleid,
         */
        tuple = SearchSysCache1(RELOID, ObjectIdGetDatum(table_oid));
        if (!HeapTupleIsValid(tuple))
-               ereport(ERROR,
-                               (errcode(ERRCODE_UNDEFINED_TABLE),
-                                errmsg("relation with OID %u does not exist",
-                                               table_oid)));
+       {
+               if (is_missing != NULL)
+               {
+                       /* return "no privileges" instead of throwing an error */
+                       *is_missing = true;
+                       return 0;
+               }
+               else
+                       ereport(ERROR,
+                                       (errcode(ERRCODE_UNDEFINED_TABLE),
+                                        errmsg("relation with OID %u does not exist",
+                                                       table_oid)));
+       }
+
        classForm = (Form_pg_class) GETSTRUCT(tuple);
 
        /*
@@ -4468,7 +4525,22 @@ AclResult
 pg_attribute_aclcheck(Oid table_oid, AttrNumber attnum,
                                          Oid roleid, AclMode mode)
 {
-       if (pg_attribute_aclmask(table_oid, attnum, roleid, mode, ACLMASK_ANY) != 0)
+       return pg_attribute_aclcheck_ext(table_oid, attnum, roleid, mode, NULL);
+}
+
+
+/*
+ * Exported routine for checking a user's access privileges to a column
+ *
+ * Does the bulk of the work for pg_attribute_aclcheck(), and allows other
+ * callers to avoid the missing attribute ERROR when is_missing is non-NULL.
+ */
+AclResult
+pg_attribute_aclcheck_ext(Oid table_oid, AttrNumber attnum,
+                                         Oid roleid, AclMode mode, bool *is_missing)
+{
+       if (pg_attribute_aclmask_ext(table_oid, attnum, roleid, mode,
+                                                                ACLMASK_ANY, is_missing) != 0)
                return ACLCHECK_OK;
        else
                return ACLCHECK_NO_PRIV;
@@ -4581,7 +4653,21 @@ pg_attribute_aclcheck_all(Oid table_oid, Oid roleid, AclMode mode,
 AclResult
 pg_class_aclcheck(Oid table_oid, Oid roleid, AclMode mode)
 {
-       if (pg_class_aclmask(table_oid, roleid, mode, ACLMASK_ANY) != 0)
+       return pg_class_aclcheck_ext(table_oid, roleid, mode, NULL);
+}
+
+/*
+ * Exported routine for checking a user's access privileges to a table
+ *
+ * Does the bulk of the work for pg_class_aclcheck(), and allows other
+ * callers to avoid the missing relation ERROR when is_missing is non-NULL.
+ */
+AclResult
+pg_class_aclcheck_ext(Oid table_oid, Oid roleid,
+                                         AclMode mode, bool *is_missing)
+{
+       if (pg_class_aclmask_ext(table_oid, roleid, mode,
+                                                        ACLMASK_ANY, is_missing) != 0)
                return ACLCHECK_OK;
        else
                return ACLCHECK_NO_PRIV;

diff --git a/src/backend/utils/adt/acl.c b/src/backend/utils/adt/acl.c
index 9955c7c5c06cdd3de2a8be3104bee32ef93f5941..6a8c6a20eeae5a828ffd059342bc0707e5449e3c 100644 (file)
--- a/src/backend/utils/adt/acl.c
+++ b/src/backend/utils/adt/acl.c
@@ -2444,8 +2444,7 @@ column_privilege_check(Oid tableoid, AttrNumber attnum,
                                           Oid roleid, AclMode mode)
 {
        AclResult       aclresult;
-       HeapTuple       attTuple;
-       Form_pg_attribute attributeForm;
+       bool            is_missing = false;
 
        /*
         * If convert_column_name failed, we can just return -1 immediately.
@@ -2454,42 +2453,25 @@ column_privilege_check(Oid tableoid, AttrNumber attnum,
                return -1;
 
        /*
-        * First check if we have the privilege at the table level.  We check
-        * existence of the pg_class row before risking calling pg_class_aclcheck.
-        * Note: it might seem there's a race condition against concurrent DROP,
-        * but really it's safe because there will be no syscache flush between
-        * here and there.  So if we see the row in the syscache, so will
-        * pg_class_aclcheck.
+        * Check for column-level privileges first. This serves in
+        * part as a check on whether the column even exists, so we
+        * need to do it before checking table-level privilege.
         */
-       if (!SearchSysCacheExists1(RELOID, ObjectIdGetDatum(tableoid)))
+       aclresult = pg_attribute_aclcheck_ext(tableoid, attnum, roleid,
+                                                                                 mode, &is_missing);
+       if (aclresult == ACLCHECK_OK)
+               return 1;
+       else if (is_missing)
                return -1;
 
-       aclresult = pg_class_aclcheck(tableoid, roleid, mode);
-
+       /* Next check if we have the privilege at the table level */
+       aclresult = pg_class_aclcheck_ext(tableoid, roleid, mode, &is_missing);
        if (aclresult == ACLCHECK_OK)
-               return true;
-
-       /*
-        * No table privilege, so try per-column privileges.  Again, we have to
-        * check for dropped attribute first, and we rely on the syscache not to
-        * notice a concurrent drop before pg_attribute_aclcheck fetches the row.
-        */
-       attTuple = SearchSysCache2(ATTNUM,
-                                                          ObjectIdGetDatum(tableoid),
-                                                          Int16GetDatum(attnum));
-       if (!HeapTupleIsValid(attTuple))
-               return -1;
-       attributeForm = (Form_pg_attribute) GETSTRUCT(attTuple);
-       if (attributeForm->attisdropped)
-       {
-               ReleaseSysCache(attTuple);
+               return 1;
+       else if (is_missing)
                return -1;
-       }
-       ReleaseSysCache(attTuple);
-
-       aclresult = pg_attribute_aclcheck(tableoid, attnum, roleid, mode);
-
-       return (aclresult == ACLCHECK_OK);
+       else
+               return 0;
 }
 
 /*

diff --git a/src/include/utils/acl.h b/src/include/utils/acl.h
index 541a438387bdf735adce76b0a83ffd8162e676e9..af771c901d1e1fc6e07c10afc290b95c3eefa669 100644 (file)
--- a/src/include/utils/acl.h
+++ b/src/include/utils/acl.h
@@ -233,8 +233,14 @@ extern void RemoveRoleFromObjectACL(Oid roleid, Oid classid, Oid objid);
 
 extern AclMode pg_attribute_aclmask(Oid table_oid, AttrNumber attnum,
                                                                        Oid roleid, AclMode mask, AclMaskHow how);
+extern AclMode pg_attribute_aclmask_ext(Oid table_oid, AttrNumber attnum,
+                                                                               Oid roleid, AclMode mask,
+                                                                               AclMaskHow how, bool *is_missing);
 extern AclMode pg_class_aclmask(Oid table_oid, Oid roleid,
                                                                AclMode mask, AclMaskHow how);
+extern AclMode pg_class_aclmask_ext(Oid table_oid, Oid roleid,
+                                                                       AclMode mask, AclMaskHow how,
+                                                                       bool *is_missing);
 extern AclMode pg_database_aclmask(Oid db_oid, Oid roleid,
                                                                   AclMode mask, AclMaskHow how);
 extern AclMode pg_proc_aclmask(Oid proc_oid, Oid roleid,
@@ -256,9 +262,14 @@ extern AclMode pg_type_aclmask(Oid type_oid, Oid roleid,
 
 extern AclResult pg_attribute_aclcheck(Oid table_oid, AttrNumber attnum,
                                                                           Oid roleid, AclMode mode);
+extern AclResult pg_attribute_aclcheck_ext(Oid table_oid, AttrNumber attnum,
+                                                                                  Oid roleid, AclMode mode,
+                                                                                  bool *is_missing);
 extern AclResult pg_attribute_aclcheck_all(Oid table_oid, Oid roleid,
                                                                                   AclMode mode, AclMaskHow how);
 extern AclResult pg_class_aclcheck(Oid table_oid, Oid roleid, AclMode mode);
+extern AclResult pg_class_aclcheck_ext(Oid table_oid, Oid roleid,
+                                                                          AclMode mode, bool *is_missing);
 extern AclResult pg_database_aclcheck(Oid db_oid, Oid roleid, AclMode mode);
 extern AclResult pg_proc_aclcheck(Oid proc_oid, Oid roleid, AclMode mode);
 extern AclResult pg_language_aclcheck(Oid lang_oid, Oid roleid, AclMode mode);

diff --git a/src/test/regress/expected/privileges.out b/src/test/regress/expected/privileges.out
index 4903371991f7f19a0c50436f252bfcc875e9682c..89f3d5da46e9dbb038960c6afa35118feae8220d 100644 (file)
--- a/src/test/regress/expected/privileges.out
+++ b/src/test/regress/expected/privileges.out
@@ -1362,7 +1362,13 @@ select has_column_privilege('mytable','........pg.dropped.2........','select');
 select has_column_privilege('mytable',2::int2,'select');
  has_column_privilege 
 ----------------------
- t
+ 
+(1 row)
+
+select has_column_privilege('mytable',99::int2,'select');
+ has_column_privilege 
+----------------------
+ 
 (1 row)
 
 revoke select on table mytable from regress_priv_user3;
@@ -1372,6 +1378,12 @@ select has_column_privilege('mytable',2::int2,'select');
  
 (1 row)
 
+select has_column_privilege('mytable',99::int2,'select');
+ has_column_privilege 
+----------------------
+ 
+(1 row)
+
 drop table mytable;
 -- Grant options
 SET SESSION AUTHORIZATION regress_priv_user1;

diff --git a/src/test/regress/sql/privileges.sql b/src/test/regress/sql/privileges.sql
index 8dcd2199e0d4dbfc449363ce94af7b5ab2d898e5..22337310af31f8eeddf24c2fdc1153475a8815f5 100644 (file)
--- a/src/test/regress/sql/privileges.sql
+++ b/src/test/regress/sql/privileges.sql
@@ -836,8 +836,10 @@ alter table mytable drop column f2;
 select has_column_privilege('mytable','f2','select');
 select has_column_privilege('mytable','........pg.dropped.2........','select');
 select has_column_privilege('mytable',2::int2,'select');
+select has_column_privilege('mytable',99::int2,'select');
 revoke select on table mytable from regress_priv_user3;
 select has_column_privilege('mytable',2::int2,'select');
+select has_column_privilege('mytable',99::int2,'select');
 drop table mytable;
 
 -- Grant options