Ensure that pg_auth_members.grantor is always valid.

Previously, "GRANT foo TO bar" or "GRANT foo TO bar GRANTED BY baz"
would record the OID of the grantor in pg_auth_members.grantor, but
that role could later be dropped without modifying or removing the
pg_auth_members record. That's not great, because we typically try
to avoid dangling references in catalog data.

Now, a role grant depends on the grantor, and the grantor can't be
dropped without removing the grant or changing the grantor.  "DROP
OWNED BY" will remove the grant, just as it does for other kinds of
privileges. "REASSIGN OWNED BY" will not, again just like what we do
in other cases involving privileges.

pg_auth_members now has an OID column, because that is needed in order
for dependencies to work. It also now has an index on the grantor
column, because otherwise dropping a role would require a sequential
scan of the entire table to see whether the role's OID is in use as
a grantor. That probably wouldn't be too large a problem in practice,
but it seems better to have an index just in case.

A follow-on patch is planned with the goal of more thoroughly
rationalizing the behavior of role grants. This patch is just trying
to do enough to make sure that the data we store in the catalogs is at
some basic level valid.

Patch by me, reviewed by Stephen Frost

Discussion: http://postgr.es/m/CA+TgmoaFr-RZeQ+WoQ5nKPv97oT9+aDgK_a5+qWHSgbDsMp1Vg@mail.gmail.com

Author: robertmhaas

doc/src/sgml/catalogs.sgml

@@ -1669,6 +1669,15 @@ SCRAM-SHA-256$<replaceable>&lt;iteration count&gt;</replaceable>:<replaceable>&l
     </thead>
 
     <tbody>
+     <row>
+      <entry role="catalog_table_entry"><para role="column_definition">
+       <structfield>oid</structfield> <type>oid</type>
+      </para>
+      <para>
+       Row identifier
+      </para></entry>
+     </row>
+
      <row>
       <entry role="catalog_table_entry"><para role="column_definition">
        <structfield>roleid</structfield> <type>oid</type>

src/backend/catalog/catalog.c

@@ -262,6 +262,8 @@ IsSharedRelation(Oid relationId)
 		relationId == AuthIdRolnameIndexId ||
 		relationId == AuthMemMemRoleIndexId ||
 		relationId == AuthMemRoleMemIndexId ||
+		relationId == AuthMemOidIndexId ||
+		relationId == AuthMemGrantorIndexId ||
 		relationId == DatabaseNameIndexId ||
 		relationId == DatabaseOidIndexId ||
 		relationId == DbRoleSettingDatidRolidIndexId ||

src/backend/catalog/dependency.c

@@ -28,6 +28,7 @@
 #include "catalog/pg_amproc.h"
 #include "catalog/pg_attrdef.h"
 #include "catalog/pg_authid.h"
+#include "catalog/pg_auth_members.h"
 #include "catalog/pg_cast.h"
 #include "catalog/pg_collation.h"
 #include "catalog/pg_constraint.h"
@@ -172,6 +173,7 @@ static const Oid object_classes[] = {
 	TSTemplateRelationId,		/* OCLASS_TSTEMPLATE */
 	TSConfigRelationId,			/* OCLASS_TSCONFIG */
 	AuthIdRelationId,			/* OCLASS_ROLE */
+	AuthMemRelationId,			/* OCLASS_ROLE_MEMBERSHIP */
 	DatabaseRelationId,			/* OCLASS_DATABASE */
 	TableSpaceRelationId,		/* OCLASS_TBLSPACE */
 	ForeignDataWrapperRelationId,	/* OCLASS_FDW */
@@ -1502,6 +1504,7 @@ doDeletion(const ObjectAddress *object, int flags)
 		case OCLASS_DEFACL:
 		case OCLASS_EVENT_TRIGGER:
 		case OCLASS_TRANSFORM:
+		case OCLASS_ROLE_MEMBERSHIP:
 			DropObjectById(object);
 			break;
 
@@ -1529,9 +1532,8 @@ doDeletion(const ObjectAddress *object, int flags)
  * Accepts the same flags as performDeletion (though currently only
  * PERFORM_DELETION_CONCURRENTLY does anything).
  *
- * We use LockRelation for relations, LockDatabaseObject for everything
- * else.  Shared-across-databases objects are not currently supported
- * because no caller cares, but could be modified to use LockSharedObject.
+ * We use LockRelation for relations, and otherwise LockSharedObject or
+ * LockDatabaseObject as appropriate for the object type.
  */
 void
 AcquireDeletionLock(const ObjectAddress *object, int flags)
@@ -1549,6 +1551,9 @@ AcquireDeletionLock(const ObjectAddress *object, int flags)
 		else
 			LockRelationOid(object->objectId, AccessExclusiveLock);
 	}
+	else if (object->classId == AuthMemRelationId)
+		LockSharedObject(object->classId, object->objectId, 0,
+						 AccessExclusiveLock);
 	else
 	{
 		/* assume we should lock the whole object not a sub-object */
@@ -2914,6 +2919,9 @@ getObjectClass(const ObjectAddress *object)
 		case AuthIdRelationId:
 			return OCLASS_ROLE;
 
+		case AuthMemRelationId:
+			return OCLASS_ROLE_MEMBERSHIP;
+
 		case DatabaseRelationId:
 			return OCLASS_DATABASE;
 

src/backend/catalog/objectaddress.c

@@ -27,6 +27,7 @@
 #include "catalog/pg_amproc.h"
 #include "catalog/pg_attrdef.h"
 #include "catalog/pg_authid.h"
+#include "catalog/pg_auth_members.h"
 #include "catalog/pg_cast.h"
 #include "catalog/pg_collation.h"
 #include "catalog/pg_constraint.h"
@@ -386,6 +387,20 @@ static const ObjectPropertyType ObjectProperty[] =
 		-1,
 		true
 	},
+	{
+		"role membership",
+		AuthMemRelationId,
+		AuthMemOidIndexId,
+		-1,
+		-1,
+		Anum_pg_auth_members_oid,
+		InvalidAttrNumber,
+		InvalidAttrNumber,
+		Anum_pg_auth_members_grantor,
+		InvalidAttrNumber,
+		-1,
+		true
+	},
 	{
 		"rule",
 		RewriteRelationId,
@@ -787,6 +802,10 @@ static const struct object_type_map
 	{
 		"role", OBJECT_ROLE
 	},
+	/* OCLASS_ROLE_MEMBERSHIP */
+	{
+		"role membership", -1	/* unmapped */
+	},
 	/* OCLASS_DATABASE */
 	{
 		"database", OBJECT_DATABASE
@@ -3644,6 +3663,48 @@ getObjectDescription(const ObjectAddress *object, bool missing_ok)
 				break;
 			}
 
+		case OCLASS_ROLE_MEMBERSHIP:
+			{
+				Relation	amDesc;
+				ScanKeyData skey[1];
+				SysScanDesc rcscan;
+				HeapTuple	tup;
+				Form_pg_auth_members amForm;
+
+				amDesc = table_open(AuthMemRelationId, AccessShareLock);
+
+				ScanKeyInit(&skey[0],
+							Anum_pg_auth_members_oid,
+							BTEqualStrategyNumber, F_OIDEQ,
+							ObjectIdGetDatum(object->objectId));
+
+				rcscan = systable_beginscan(amDesc, AuthMemOidIndexId, true,
+											NULL, 1, skey);
+
+				tup = systable_getnext(rcscan);
+
+				if (!HeapTupleIsValid(tup))
+				{
+					if (!missing_ok)
+						elog(ERROR, "could not find tuple for role membership %u",
+							 object->objectId);
+
+					systable_endscan(rcscan);
+					table_close(amDesc, AccessShareLock);
+					break;
+				}
+
+				amForm = (Form_pg_auth_members) GETSTRUCT(tup);
+
+				appendStringInfo(&buffer, _("membership of role %s in role %s"),
+								 GetUserNameFromId(amForm->member, false),
+								 GetUserNameFromId(amForm->roleid, false));
+
+				systable_endscan(rcscan);
+				table_close(amDesc, AccessShareLock);
+				break;
+			}
+
 		case OCLASS_DATABASE:
 			{
 				char	   *datname;
@@ -4533,6 +4594,10 @@ getObjectTypeDescription(const ObjectAddress *object, bool missing_ok)
 			appendStringInfoString(&buffer, "role");
 			break;
 
+		case OCLASS_ROLE_MEMBERSHIP:
+			appendStringInfoString(&buffer, "role membership");
+			break;
+
 		case OCLASS_DATABASE:
 			appendStringInfoString(&buffer, "database");
 			break;
@@ -5476,6 +5541,49 @@ getObjectIdentityParts(const ObjectAddress *object,
 				break;
 			}
 
+		case OCLASS_ROLE_MEMBERSHIP:
+			{
+				Relation	authMemDesc;
+				ScanKeyData skey[1];
+				SysScanDesc amscan;
+				HeapTuple	tup;
+				Form_pg_auth_members amForm;
+
+				authMemDesc = table_open(AuthMemRelationId,
+										 AccessShareLock);
+
+				ScanKeyInit(&skey[0],
+							Anum_pg_auth_members_oid,
+							BTEqualStrategyNumber, F_OIDEQ,
+							ObjectIdGetDatum(object->objectId));
+
+				amscan = systable_beginscan(authMemDesc, AuthMemOidIndexId, true,
+											NULL, 1, skey);
+
+				tup = systable_getnext(amscan);
+
+				if (!HeapTupleIsValid(tup))
+				{
+					if (!missing_ok)
+						elog(ERROR, "could not find tuple for pg_auth_members entry %u",
+							 object->objectId);
+
+					systable_endscan(amscan);
+					table_close(authMemDesc, AccessShareLock);
+					break;
+				}
+
+				amForm = (Form_pg_auth_members) GETSTRUCT(tup);
+
+				appendStringInfo(&buffer, _("membership of role %s in role %s"),
+								 GetUserNameFromId(amForm->member, false),
+								 GetUserNameFromId(amForm->roleid, false));
+
+				systable_endscan(amscan);
+				table_close(authMemDesc, AccessShareLock);
+				break;
+			}
+
 		case OCLASS_DATABASE:
 			{
 				char	   *datname;

src/backend/catalog/pg_shdepend.c

@@ -22,6 +22,7 @@
 #include "catalog/dependency.h"
 #include "catalog/indexing.h"
 #include "catalog/pg_authid.h"
+#include "catalog/pg_auth_members.h"
 #include "catalog/pg_collation.h"
 #include "catalog/pg_conversion.h"
 #include "catalog/pg_database.h"
@@ -1364,11 +1365,6 @@ shdepDropOwned(List *roleids, DropBehavior behavior)
 				case SHARED_DEPENDENCY_INVALID:
 					elog(ERROR, "unexpected dependency type");
 					break;
-				case SHARED_DEPENDENCY_ACL:
-					RemoveRoleFromObjectACL(roleid,
-											sdepForm->classid,
-											sdepForm->objid);
-					break;
 				case SHARED_DEPENDENCY_POLICY:
 
 					/*
@@ -1398,22 +1394,37 @@ shdepDropOwned(List *roleids, DropBehavior behavior)
 						add_exact_object_address(&obj, deleteobjs);
 					}
 					break;
+				case SHARED_DEPENDENCY_ACL:
+
+					/*
+					 * Dependencies on role grants are recorded using
+					 * SHARED_DEPENDENCY_ACL, but unlike a regular ACL list
+					 * which stores all permissions for a particular object in
+					 * a single ACL array, there's a separate catalog row for
+					 * each grant - so removing the grant just means removing
+					 * the entire row.
+					 */
+					if (sdepForm->classid != AuthMemRelationId)
+					{
+						RemoveRoleFromObjectACL(roleid,
+												sdepForm->classid,
+												sdepForm->objid);
+						break;
+					}
+					/* FALLTHROUGH */
 				case SHARED_DEPENDENCY_OWNER:
-					/* If a local object, save it for deletion below */
-					if (sdepForm->dbid == MyDatabaseId)
+					/* Save it for deletion below */
+					obj.classId = sdepForm->classid;
+					obj.objectId = sdepForm->objid;
+					obj.objectSubId = sdepForm->objsubid;
+					/* as above */
+					AcquireDeletionLock(&obj, 0);
+					if (!systable_recheck_tuple(scan, tuple))
 					{
-						obj.classId = sdepForm->classid;
-						obj.objectId = sdepForm->objid;
-						obj.objectSubId = sdepForm->objsubid;
-						/* as above */
-						AcquireDeletionLock(&obj, 0);
-						if (!systable_recheck_tuple(scan, tuple))
-						{
-							ReleaseDeletionLock(&obj);
-							break;
-						}
-						add_exact_object_address(&obj, deleteobjs);
+						ReleaseDeletionLock(&obj);
+						break;
 					}
+					add_exact_object_address(&obj, deleteobjs);
 					break;
 			}
 		}

src/backend/catalog/system_views.sql

@@ -53,7 +53,7 @@ CREATE VIEW pg_group AS
     SELECT
         rolname AS groname,
         oid AS grosysid,
-        ARRAY(SELECT member FROM pg_auth_members WHERE roleid = oid) AS grolist
+        ARRAY(SELECT member FROM pg_auth_members WHERE roleid = pg_authid.oid) AS grolist
     FROM pg_authid
     WHERE NOT rolcanlogin;
 

src/backend/commands/alter.c

@@ -650,6 +650,7 @@ AlterObjectNamespace_oid(Oid classId, Oid objid, Oid nspOid,
 		case OCLASS_TRIGGER:
 		case OCLASS_SCHEMA:
 		case OCLASS_ROLE:
+		case OCLASS_ROLE_MEMBERSHIP:
 		case OCLASS_DATABASE:
 		case OCLASS_TBLSPACE:
 		case OCLASS_FDW:

src/backend/commands/event_trigger.c

@@ -1015,6 +1015,7 @@ EventTriggerSupportsObjectClass(ObjectClass objclass)
 		case OCLASS_DATABASE:
 		case OCLASS_TBLSPACE:
 		case OCLASS_ROLE:
+		case OCLASS_ROLE_MEMBERSHIP:
 		case OCLASS_PARAMETER_ACL:
 			/* no support for global objects */
 			return false;

src/backend/commands/tablecmds.c

@@ -12667,6 +12667,7 @@ ATExecAlterColumnType(AlteredTableInfo *tab, Relation rel,
 			case OCLASS_TSTEMPLATE:
 			case OCLASS_TSCONFIG:
 			case OCLASS_ROLE:
+			case OCLASS_ROLE_MEMBERSHIP:
 			case OCLASS_DATABASE:
 			case OCLASS_TBLSPACE:
 			case OCLASS_FDW: