Subject: [PATCH] Draft implementation of USAGE privilege on PUBLICATION.

From: Antonin Houska <[email-masked]>
Date: Wed, 18 May 2022 10:58:42 +0200

This privilege seems to be useful for databases from which multiple
subscribers replicate the data. It should not be assumed that any publication
can be exposed to any subscription.
---
 doc/src/sgml/ddl.sgml                       |   7 +
 doc/src/sgml/logical-replication.sgml       |   4 +-
 doc/src/sgml/ref/grant.sgml                 |   7 +-
 doc/src/sgml/ref/revoke.sgml                |   7 +
 src/backend/catalog/aclchk.c                | 206 ++++++++++++++++++++
 src/backend/commands/copyto.c               | 114 +++++++++++
 src/backend/commands/publicationcmds.c      |   2 +
 src/backend/parser/gram.y                   |   8 +
 src/backend/replication/pgoutput/pgoutput.c |  13 +-
 src/backend/utils/adt/acl.c                 |   7 +
 src/bin/pg_dump/dumputils.c                 |   2 +
 src/bin/pg_dump/pg_dump.c                   |  28 ++-
 src/bin/pg_dump/pg_dump.h                   |   1 +
 src/bin/psql/tab-complete.c                 |   3 +
 src/include/catalog/pg_publication.h        |   6 +
 src/include/nodes/parsenodes.h              |   4 +-
 src/include/utils/acl.h                     |   4 +
 17 files changed, 413 insertions(+), 10 deletions(-)

diff --git a/doc/src/sgml/ddl.sgml b/doc/src/sgml/ddl.sgml
index f2ac1ba0034..dc499c50756 100644
--- a/doc/src/sgml/ddl.sgml
+++ b/doc/src/sgml/ddl.sgml
@@ -1935,6 +1935,13 @@ REVOKE ALL ON accounts FROM PUBLIC;
        statements that have previously performed this lookup, so this is not
        a completely secure way to prevent object access.
       </para>
+      <para>
+       For publications, allows logical replication via particular
+       publication. The user specified in
+       the <link linkend="sql-createsubscription"><command>CREATE
+       SUBSCRIPTION</command></link> command must have this privilege on all
+       publications listed in that command.
+      </para>
       <para>
        For sequences, allows use of the
        <function>currval</function> and <function>nextval</function> functions.
diff --git a/doc/src/sgml/logical-replication.sgml b/doc/src/sgml/logical-replication.sgml
index 145ea71d61b..63194b34541 100644
--- a/doc/src/sgml/logical-replication.sgml
+++ b/doc/src/sgml/logical-replication.sgml
@@ -1156,7 +1156,9 @@ CONTEXT:  processing remote data for replication origin "pg_16395" during "INSER
   <para>
    In order to be able to copy the initial table data, the role used for the
    replication connection must have the <literal>SELECT</literal> privilege on
-   a published table (or be a superuser).
+   a published table (or be a superuser). In addition, the role must have
+   the <literal>USAGE</literal> privilege on all the publications referenced
+   by particular subscription.
   </para>
 
   <para>
diff --git a/doc/src/sgml/ref/grant.sgml b/doc/src/sgml/ref/grant.sgml
index f744b05b55d..95b9c46137f 100644
--- a/doc/src/sgml/ref/grant.sgml
+++ b/doc/src/sgml/ref/grant.sgml
@@ -82,6 +82,11 @@ GRANT { { SET | ALTER SYSTEM } [, ... ] | ALL [ PRIVILEGES ] }
     TO <replaceable class="parameter">role_specification</replaceable> [, ...] [ WITH GRANT OPTION ]
     [ GRANTED BY <replaceable class="parameter">role_specification</replaceable> ]
 
+GRANT { USAGE | ALL [ PRIVILEGES ] }
+    ON PUBLICATION <replaceable>pub_name</replaceable> [, ...]
+    TO <replaceable class="parameter">role_specification</replaceable> [, ...] [ WITH GRANT OPTION ]
+    [ GRANTED BY <replaceable class="parameter">role_specification</replaceable> ]
+
 GRANT { { CREATE | USAGE } [, ...] | ALL [ PRIVILEGES ] }
     ON SCHEMA <replaceable>schema_name</replaceable> [, ...]
     TO <replaceable class="parameter">role_specification</replaceable> [, ...] [ WITH GRANT OPTION ]
@@ -460,7 +465,7 @@ GRANT admins TO joe;
    </para>
 
    <para>
-    Privileges on databases, tablespaces, schemas, languages, and
+    Privileges on databases, tablespaces, schemas, languages, publications and
     configuration parameters are
     <productname>PostgreSQL</productname> extensions.
    </para>
diff --git a/doc/src/sgml/ref/revoke.sgml b/doc/src/sgml/ref/revoke.sgml
index 62f19710369..c21ea4c9c69 100644
--- a/doc/src/sgml/ref/revoke.sgml
+++ b/doc/src/sgml/ref/revoke.sgml
@@ -104,6 +104,13 @@ REVOKE [ GRANT OPTION FOR ]
     [ GRANTED BY <replaceable class="parameter">role_specification</replaceable> ]
     [ CASCADE | RESTRICT ]
 
+REVOKE [ GRANT OPTION FOR ]
+    { USAGE | ALL [ PRIVILEGES ] }
+    ON PUBLICATION <replaceable>pub_name</replaceable> [, ...]
+    FROM <replaceable class="parameter">role_specification</replaceable> [, ...]
+    [ GRANTED BY <replaceable class="parameter">role_specification</replaceable> ]
+    [ CASCADE | RESTRICT ]
+
 REVOKE [ GRANT OPTION FOR ]
     { { CREATE | USAGE } [, ...] | ALL [ PRIVILEGES ] }
     ON SCHEMA <replaceable>schema_name</replaceable> [, ...]
diff --git a/src/backend/catalog/aclchk.c b/src/backend/catalog/aclchk.c
index 5f1726c0957..2896c10415f 100644
--- a/src/backend/catalog/aclchk.c
+++ b/src/backend/catalog/aclchk.c
@@ -110,6 +110,7 @@ static void ExecGrant_ForeignServer(InternalGrant *grantStmt);
 static void ExecGrant_Function(InternalGrant *grantStmt);
 static void ExecGrant_Language(InternalGrant *grantStmt);
 static void ExecGrant_Largeobject(InternalGrant *grantStmt);
+static void ExecGrant_Publication(InternalGrant *istmt);
 static void ExecGrant_Namespace(InternalGrant *grantStmt);
 static void ExecGrant_Tablespace(InternalGrant *grantStmt);
 static void ExecGrant_Type(InternalGrant *grantStmt);
@@ -477,6 +478,10 @@ ExecuteGrantStmt(GrantStmt *stmt)
 			all_privileges = ACL_ALL_RIGHTS_LARGEOBJECT;
 			errormsg = gettext_noop("invalid privilege type %s for large object");
 			break;
+		case OBJECT_PUBLICATION:
+			all_privileges = ACL_ALL_RIGHTS_PUBLICATION;
+			errormsg = gettext_noop("invalid privilege type %s for publication");
+			break;
 		case OBJECT_SCHEMA:
 			all_privileges = ACL_ALL_RIGHTS_SCHEMA;
 			errormsg = gettext_noop("invalid privilege type %s for schema");
@@ -605,6 +610,9 @@ ExecGrantStmt_oids(InternalGrant *istmt)
 		case OBJECT_LARGEOBJECT:
 			ExecGrant_Largeobject(istmt);
 			break;
+		case OBJECT_PUBLICATION:
+			ExecGrant_Publication(istmt);
+			break;
 		case OBJECT_SCHEMA:
 			ExecGrant_Namespace(istmt);
 			break;
@@ -715,6 +723,16 @@ objectNamesToOids(ObjectType objtype, List *objnames, bool is_grant)
 				objects = lappend_oid(objects, lobjOid);
 			}
 			break;
+		case OBJECT_PUBLICATION:
+			foreach(cell, objnames)
+			{
+				char	   *nspname = strVal(lfirst(cell));
+				Oid			oid;
+
+				oid = get_publication_oid(nspname, false);
+				objects = lappend_oid(objects, oid);
+			}
+			break;
 		case OBJECT_SCHEMA:
 			foreach(cell, objnames)
 			{
@@ -2892,6 +2910,126 @@ ExecGrant_Largeobject(InternalGrant *istmt)
 	table_close(relation, RowExclusiveLock);
 }
 
+static void
+ExecGrant_Publication(InternalGrant *istmt)
+{
+	Relation	relation;
+	ListCell   *cell;
+
+	if (istmt->all_privs && istmt->privileges == ACL_NO_RIGHTS)
+		istmt->privileges = ACL_ALL_RIGHTS_PUBLICATION;
+
+	relation = table_open(PublicationRelationId, RowExclusiveLock);
+
+	foreach(cell, istmt->objects)
+	{
+		Oid			pubid = lfirst_oid(cell);
+		HeapTuple	tuple;
+		HeapTuple	newtuple;
+		Form_pg_publication pg_pub_tuple;
+		Oid			ownerId;
+		Datum		aclDatum;
+		bool		isnull;
+		AclMode		avail_goptions;
+		AclMode		this_privileges;
+		Acl		   *old_acl;
+		Acl		   *new_acl;
+		Oid			grantorId;
+		Datum		values[Natts_pg_publication];
+		bool		nulls[Natts_pg_publication];
+		bool		replaces[Natts_pg_publication];
+		int			noldmembers;
+		int			nnewmembers;
+		Oid		   *oldmembers;
+		Oid		   *newmembers;
+
+		tuple = SearchSysCache1(PUBLICATIONOID, ObjectIdGetDatum(pubid));
+		if (!HeapTupleIsValid(tuple))
+			elog(ERROR, "cache lookup failed for publication %u", pubid);
+
+		pg_pub_tuple = (Form_pg_publication) GETSTRUCT(tuple);
+		ownerId = pg_pub_tuple->pubowner;
+		aclDatum = SysCacheGetAttr(PUBLICATIONOID, tuple,
+								   Anum_pg_publication_pubacl,
+								   &isnull);
+
+		if (isnull)
+		{
+			old_acl = acldefault(OBJECT_PUBLICATION, ownerId);
+			/* There are no old member roles according to the catalogs */
+			noldmembers = 0;
+			oldmembers = NULL;
+		}
+		else
+		{
+			old_acl = DatumGetAclPCopy(aclDatum);
+			/* Get the roles mentioned in the existing ACL */
+			noldmembers = aclmembers(old_acl, &oldmembers);
+		}
+
+		/* Determine ID to do the grant as, and available grant options */
+		select_best_grantor(GetUserId(), istmt->privileges,
+							old_acl, ownerId,
+							&grantorId, &avail_goptions);
+
+		/*
+		 * Restrict the privileges to what we can actually grant, and emit the
+		 * standards-mandated warning and error messages.
+		 */
+		this_privileges =
+			restrict_and_check_grant(istmt->is_grant, avail_goptions,
+									 istmt->all_privs, istmt->privileges,
+									 pubid, grantorId, OBJECT_FUNCTION,
+									 NameStr(pg_pub_tuple->pubname),
+									 0, NULL);
+
+		/*
+		 * Generate new ACL.
+		 */
+		new_acl = merge_acl_with_grant(old_acl, istmt->is_grant,
+									   istmt->grant_option, istmt->behavior,
+									   istmt->grantees, this_privileges,
+									   grantorId, ownerId);
+
+		/*
+		 * We need the members of both old and new ACLs so we can correct the
+		 * shared dependency information.
+		 */
+		nnewmembers = aclmembers(new_acl, &newmembers);
+
+		/* finished building new ACL value, now insert it */
+		MemSet(values, 0, sizeof(values));
+		MemSet(nulls, false, sizeof(nulls));
+		MemSet(replaces, false, sizeof(replaces));
+
+		replaces[Anum_pg_publication_pubacl - 1] = true;
+		values[Anum_pg_publication_pubacl - 1] = PointerGetDatum(new_acl);
+
+		newtuple = heap_modify_tuple(tuple, RelationGetDescr(relation), values,
+									 nulls, replaces);
+
+		CatalogTupleUpdate(relation, &newtuple->t_self, newtuple);
+
+		/* Update initial privileges for extensions */
+		recordExtensionInitPriv(pubid, PublicationRelationId, 0, new_acl);
+
+		/* Update the shared dependency ACL info */
+		updateAclDependencies(PublicationRelationId, pubid, 0,
+							  ownerId,
+							  noldmembers, oldmembers,
+							  nnewmembers, newmembers);
+
+		ReleaseSysCache(tuple);
+
+		pfree(new_acl);
+
+		/* prevent error when processing duplicate objects */
+		CommandCounterIncrement();
+	}
+
+	table_close(relation, RowExclusiveLock);
+}
+
 static void
 ExecGrant_Namespace(InternalGrant *istmt)
 {
@@ -3856,6 +3994,8 @@ pg_aclmask(ObjectType objtype, Oid table_oid, AttrNumber attnum, Oid roleid,
 			return pg_database_aclmask(table_oid, roleid, mask, how);
 		case OBJECT_FUNCTION:
 			return pg_proc_aclmask(table_oid, roleid, mask, how);
+		case OBJECT_PUBLICATION:
+			return pg_publication_aclmask(table_oid, roleid, mask, how);
 		case OBJECT_LANGUAGE:
 			return pg_language_aclmask(table_oid, roleid, mask, how);
 		case OBJECT_LARGEOBJECT:
@@ -4379,6 +4519,60 @@ pg_proc_aclmask(Oid proc_oid, Oid roleid,
 	return result;
 }
 
+/*
+ * Exported routine for examining a user's privileges for a publication.
+ */
+AclMode
+pg_publication_aclmask(Oid pubid, Oid roleid, AclMode mask,
+					   AclMaskHow how)
+{
+	AclMode		result;
+	HeapTuple	tuple;
+	Datum		aclDatum;
+	bool		isNull;
+	Acl		   *acl;
+	Oid			ownerId;
+
+	/* Superusers bypass all permission checking. */
+	if (superuser_arg(roleid))
+		return mask;
+
+	/*
+	 * Get the publication's ACL from pg_publication
+	 */
+	tuple = SearchSysCache1(PUBLICATIONOID, ObjectIdGetDatum(pubid));
+	if (!HeapTupleIsValid(tuple))
+		ereport(ERROR,
+				(errcode(ERRCODE_UNDEFINED_FUNCTION),
+				 errmsg("publication with OID %u does not exist", pubid)));
+
+	ownerId = ((Form_pg_publication) GETSTRUCT(tuple))->pubowner;
+
+	aclDatum = SysCacheGetAttr(PUBLICATIONOID, tuple, Anum_pg_publication_pubacl,
+							   &isNull);
+	if (isNull)
+	{
+		/* No ACL, so build default ACL */
+		acl = acldefault(OBJECT_PUBLICATION, ownerId);
+		aclDatum = (Datum) 0;
+	}
+	else
+	{
+		/* detoast ACL if necessary */
+		acl = DatumGetAclP(aclDatum);
+	}
+
+	result = aclmask(acl, roleid, ownerId, mask, how);
+
+	/* if we have a detoasted copy, free it */
+	if (acl && (Pointer) acl != DatumGetPointer(aclDatum))
+		pfree(acl);
+
+	ReleaseSysCache(tuple);
+
+	return result;
+}
+
 /*
  * Exported routine for examining a user's privileges for a language
  */
@@ -5076,6 +5270,18 @@ pg_proc_aclcheck(Oid proc_oid, Oid roleid, AclMode mode)
 		return ACLCHECK_NO_PRIV;
 }
 
+/*
+ * Exported routine for checking a user's access privileges to a publication
+ */
+AclResult
+pg_publication_aclcheck(Oid pub_oid, Oid roleid, AclMode mode)
+{
+	if (pg_publication_aclmask(pub_oid, roleid, mode, ACLMASK_ANY) != 0)
+		return ACLCHECK_OK;
+	else
+		return ACLCHECK_NO_PRIV;
+}
+
 /*
  * Exported routine for checking a user's access privileges to a language
  */
diff --git a/src/backend/commands/copyto.c b/src/backend/commands/copyto.c
index fca29a9a105..0d39ef77610 100644
--- a/src/backend/commands/copyto.c
+++ b/src/backend/commands/copyto.c
@@ -27,6 +27,7 @@
 #include "commands/progress.h"
 #include "executor/execdesc.h"
 #include "executor/executor.h"
+#include "executor/spi.h"
 #include "executor/tuptable.h"
 #include "libpq/libpq.h"
 #include "libpq/pqformat.h"
@@ -34,9 +35,11 @@
 #include "miscadmin.h"
 #include "optimizer/optimizer.h"
 #include "pgstat.h"
+#include "replication/walsender.h"
 #include "rewrite/rewriteHandler.h"
 #include "storage/fd.h"
 #include "tcop/tcopprot.h"
+#include "utils/acl.h"
 #include "utils/lsyscache.h"
 #include "utils/memutils.h"
 #include "utils/partcache.h"
@@ -129,6 +132,7 @@ static void CopySendEndOfRow(CopyToState cstate);
 static void CopySendInt32(CopyToState cstate, int32 val);
 static void CopySendInt16(CopyToState cstate, int16 val);
 
+static void check_publication_privileges(List *relids);
 
 /*
  * Send copy start/stop messages for frontend copies.  These have changed
@@ -360,6 +364,7 @@ BeginCopyTo(ParseState *pstate,
 		PROGRESS_COPY_COMMAND_TO,
 		0
 	};
+	List	*pubrelids = NIL;
 
 	if (rel != NULL && rel->rd_rel->relkind != RELKIND_RELATION)
 	{
@@ -424,6 +429,13 @@ BeginCopyTo(ParseState *pstate,
 		cstate->rel = rel;
 
 		tupDesc = RelationGetDescr(cstate->rel);
+
+		/*
+		 * In walsender, we need to check the privileges of the related
+		 * publications.
+		 */
+		if (am_walsender)
+			pubrelids = lappend_oid(pubrelids, RelationGetRelid(rel));
 	}
 	else
 	{
@@ -553,9 +565,32 @@ BeginCopyTo(ParseState *pstate,
 		 */
 		ExecutorStart(cstate->queryDesc, 0);
 
+		/*
+		 * In walsender, we need to check the privileges of the related
+		 * publications.
+		 */
+		if (am_walsender)
+		{
+			ListCell	*lc;
+
+			foreach(lc, plan->rtable)
+			{
+				RangeTblEntry	*rte = lfirst_node(RangeTblEntry, lc);
+
+				if (rte->rtekind != RTE_RELATION)
+					continue;
+
+				pubrelids = lappend_oid(pubrelids, rte->relid);
+			}
+		}
+
 		tupDesc = cstate->queryDesc->tupDesc;
 	}
 
+	/* Check if publication privileges allow reading from the relations. */
+	if (pubrelids)
+		check_publication_privileges(pubrelids);
+
 	/* Generate or convert list of attributes to process */
 	cstate->attnumlist = CopyGetAttnums(tupDesc, cstate->rel, attnamelist);
 
@@ -1308,3 +1343,82 @@ CreateCopyDestReceiver(void)
 
 	return (DestReceiver *) self;
 }
+
+static void
+check_publication_privileges(List *relids)
+{
+	ListCell	*lc;
+	StringInfoData	bufOids;
+	StringInfoData	bufQuery;
+	int		i;
+	List	*pubids = NIL;
+
+	if (list_length(relids) == 0)
+		return;
+
+	/* Construct a list of the OIDs */
+	initStringInfo(&bufOids);
+	foreach(lc, relids)
+	{
+		Oid		oid = lfirst_oid(lc);
+
+		if (bufOids.len > 0)
+			appendStringInfoString(&bufOids, ", ");
+		appendStringInfo(&bufOids, "%u", oid);
+	}
+
+	/*
+	 * Construct the query to retrieve the publication names. (Similar query
+	 * is in fetch_remote_table_info()).
+	 */
+	initStringInfo(&bufQuery);
+	appendStringInfo(&bufQuery,
+					 "SELECT DISTINCT p.oid"
+					 "  FROM pg_publication p"
+					 "  LEFT OUTER JOIN pg_publication_rel pr"
+					 "       ON p.oid = pr.prpubid, "
+					 "  LATERAL pg_get_publication_tables(p.pubname) gpt"
+					 " WHERE pr.prrelid IN (%s) AND gpt.relid IN (%s)",
+					 bufOids.data, bufOids.data);
+	list_free(relids);
+
+	/* Run the query to get the publication names. */
+	if (SPI_connect() != SPI_OK_CONNECT)
+		elog(ERROR, "SPI_connect failed");
+
+	if (SPI_exec(bufQuery.data, 0) != SPI_OK_SELECT)
+		elog(ERROR, "SPI_exec failed: %s", bufQuery.data);
+
+	for (i = 0; i < SPI_processed; i++)
+	{
+		Datum	oidD;
+		bool	isnull;
+
+		oidD = SPI_getbinval(SPI_tuptable->vals[i],
+							 SPI_tuptable->tupdesc,
+							 1,
+							 &isnull);
+		Assert(!isnull);
+		pubids = lappend_oid(pubids, DatumGetObjectId(oidD));
+	}
+
+	/* Finally, check the privileges. */
+	foreach(lc, pubids)
+	{
+		Oid		oid = lfirst_oid(lc);
+
+		if (pg_publication_aclcheck(oid, GetUserId(), ACL_USAGE) != ACLCHECK_OK)
+		{
+			char	*pubname = get_publication_name(oid, false);
+
+			ereport(ERROR,
+					(errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
+					 errmsg("permission denied for publication %s",
+							pubname)));
+		}
+	}
+	list_free(pubids);
+
+	if (SPI_finish() != SPI_OK_FINISH)
+		elog(ERROR, "SPI_finish failed");
+}
diff --git a/src/backend/commands/publicationcmds.c b/src/backend/commands/publicationcmds.c
index 8e645741e4e..bb7eb5726c4 100644
--- a/src/backend/commands/publicationcmds.c
+++ b/src/backend/commands/publicationcmds.c
@@ -822,6 +822,8 @@ CreatePublication(ParseState *pstate, CreatePublicationStmt *stmt)
 		BoolGetDatum(pubactions.pubtruncate);
 	values[Anum_pg_publication_pubviaroot - 1] =
 		BoolGetDatum(publish_via_partition_root);
+	values[Anum_pg_publication_pubowner - 1] = ObjectIdGetDatum(GetUserId());
+	nulls[Anum_pg_publication_pubacl - 1] = true;
 
 	tup = heap_form_tuple(RelationGetDescr(rel), values, nulls);
 
diff --git a/src/backend/parser/gram.y b/src/backend/parser/gram.y
index 989db0dbece..3b8dffc72ee 100644
--- a/src/backend/parser/gram.y
+++ b/src/backend/parser/gram.y
@@ -7720,6 +7720,14 @@ privilege_target:
 					n->objs = $2;
 					$$ = n;
 				}
+			| PUBLICATION name_list
+				{
+					PrivTarget *n = (PrivTarget *) palloc(sizeof(PrivTarget));
+					n->targtype = ACL_TARGET_OBJECT;
+					n->objtype = OBJECT_PUBLICATION;
+					n->objs = $2;
+					$$ = n;
+				}
 			| SCHEMA name_list
 				{
 					PrivTarget *n = (PrivTarget *) palloc(sizeof(PrivTarget));
diff --git a/src/backend/replication/pgoutput/pgoutput.c b/src/backend/replication/pgoutput/pgoutput.c
index 42c06af2391..d15969252da 100644
--- a/src/backend/replication/pgoutput/pgoutput.c
+++ b/src/backend/replication/pgoutput/pgoutput.c
@@ -19,12 +19,14 @@
 #include "commands/defrem.h"
 #include "executor/executor.h"
 #include "fmgr.h"
+#include "miscadmin.h"
 #include "nodes/makefuncs.h"
 #include "optimizer/optimizer.h"
 #include "replication/logical.h"
 #include "replication/logicalproto.h"
 #include "replication/origin.h"
 #include "replication/pgoutput.h"
+#include "utils/acl.h"
 #include "utils/builtins.h"
 #include "utils/inval.h"
 #include "utils/lsyscache.h"
@@ -2082,8 +2084,15 @@ get_rel_sync_entry(PGOutputData *data, Relation relation)
 			 * We'll use the top-most relid across all publications. Also
 			 * track the ancestor level for this publication.
 			 */
-			Oid			pub_relid = relid;
-			int			ancestor_level = 0;
+			Oid	pub_relid = relid;
+			int	ancestor_level = 0;
+
+			/* Check if we have the usage privilege on the publication. */
+			if (pg_publication_aclcheck(pub->oid, GetUserId(), ACL_USAGE) != ACLCHECK_OK)
+				ereport(ERROR,
+						(errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
+						 errmsg("permission denied for publication %s",
+								pub->name)));
 
 			/*
 			 * If this is a FOR ALL TABLES publication, pick the partition
diff --git a/src/backend/utils/adt/acl.c b/src/backend/utils/adt/acl.c
index 772c04155c3..fc3fd26a751 100644
--- a/src/backend/utils/adt/acl.c
+++ b/src/backend/utils/adt/acl.c
@@ -807,6 +807,10 @@ acldefault(ObjectType objtype, Oid ownerId)
 			world_default = ACL_NO_RIGHTS;
 			owner_default = ACL_ALL_RIGHTS_PARAMETER_ACL;
 			break;
+		case OBJECT_PUBLICATION:
+			world_default = ACL_USAGE;
+			owner_default = ACL_ALL_RIGHTS_PUBLICATION;
+			break;
 		default:
 			elog(ERROR, "unrecognized objtype: %d", (int) objtype);
 			world_default = ACL_NO_RIGHTS;	/* keep compiler quiet */
@@ -904,6 +908,9 @@ acldefault_sql(PG_FUNCTION_ARGS)
 		case 'T':
 			objtype = OBJECT_TYPE;
 			break;
+		case 'P':
+			objtype = OBJECT_PUBLICATION;
+			break;
 		default:
 			elog(ERROR, "unrecognized objtype abbreviation: %c", objtypec);
 	}
diff --git a/src/bin/pg_dump/dumputils.c b/src/bin/pg_dump/dumputils.c
index 3e68dfc78f9..4c5b6d095ff 100644
--- a/src/bin/pg_dump/dumputils.c
+++ b/src/bin/pg_dump/dumputils.c
@@ -511,6 +511,8 @@ do { \
 		CONVERT_PRIV('r', "SELECT");
 		CONVERT_PRIV('w', "UPDATE");
 	}
+	else if (strcmp(type, "PUBLICATION") == 0)
+		CONVERT_PRIV('U', "USAGE");
 	else
 		abort();
 
diff --git a/src/bin/pg_dump/pg_dump.c b/src/bin/pg_dump/pg_dump.c
index 7cc9c72e492..7deae290507 100644
--- a/src/bin/pg_dump/pg_dump.c
+++ b/src/bin/pg_dump/pg_dump.c
@@ -3868,6 +3868,8 @@ getPublications(Archive *fout, int *numPublications)
 	int			i_pubdelete;
 	int			i_pubtruncate;
 	int			i_pubviaroot;
+	int			i_pubacl;
+	int			i_acldefault;
 	int			i,
 				ntups;
 
@@ -3882,23 +3884,29 @@ getPublications(Archive *fout, int *numPublications)
 	resetPQExpBuffer(query);
 
 	/* Get the publications. */
-	if (fout->remoteVersion >= 130000)
+	if (fout->remoteVersion >= 150000)
 		appendPQExpBuffer(query,
 						  "SELECT p.tableoid, p.oid, p.pubname, "
 						  "p.pubowner, "
-						  "p.puballtables, p.pubinsert, p.pubupdate, p.pubdelete, p.pubtruncate, p.pubviaroot "
+						  "p.puballtables, p.pubinsert, p.pubupdate, p.pubdelete, p.pubtruncate, p.pubviaroot, p.pubacl, acldefault('P', p.pubowner) AS acldefault "
+						  "FROM pg_publication p");
+	else if (fout->remoteVersion >= 130000)
+		appendPQExpBuffer(query,
+						  "SELECT p.tableoid, p.oid, p.pubname, "
+						  "p.pubowner, "
+						  "p.puballtables, p.pubinsert, p.pubupdate, p.pubdelete, p.pubtruncate, p.pubviaroot, '{}' AS pubacl, '{}' AS acldefault "
 						  "FROM pg_publication p");
 	else if (fout->remoteVersion >= 110000)
 		appendPQExpBuffer(query,
 						  "SELECT p.tableoid, p.oid, p.pubname, "
 						  "p.pubowner, "
-						  "p.puballtables, p.pubinsert, p.pubupdate, p.pubdelete, p.pubtruncate, false AS pubviaroot "
+						  "p.puballtables, p.pubinsert, p.pubupdate, p.pubdelete, p.pubtruncate, false AS pubviaroot, '{}' AS pubacl, '{}' AS acldefault "
 						  "FROM pg_publication p");
 	else
 		appendPQExpBuffer(query,
 						  "SELECT p.tableoid, p.oid, p.pubname, "
 						  "p.pubowner, "
-						  "p.puballtables, p.pubinsert, p.pubupdate, p.pubdelete, false AS pubtruncate, false AS pubviaroot "
+						  "p.puballtables, p.pubinsert, p.pubupdate, p.pubdelete, false AS pubtruncate, false AS pubviaroot, '{}' AS pubacl, '{}' AS acldefault "
 						  "FROM pg_publication p");
 
 	res = ExecuteSqlQuery(fout, query->data, PGRES_TUPLES_OK);
@@ -3915,6 +3923,8 @@ getPublications(Archive *fout, int *numPublications)
 	i_pubdelete = PQfnumber(res, "pubdelete");
 	i_pubtruncate = PQfnumber(res, "pubtruncate");
 	i_pubviaroot = PQfnumber(res, "pubviaroot");
+	i_pubacl = PQfnumber(res, "pubacl");
+	i_acldefault = PQfnumber(res, "acldefault");
 
 	pubinfo = pg_malloc(ntups * sizeof(PublicationInfo));
 
@@ -3939,6 +3949,11 @@ getPublications(Archive *fout, int *numPublications)
 			(strcmp(PQgetvalue(res, i, i_pubtruncate), "t") == 0);
 		pubinfo[i].pubviaroot =
 			(strcmp(PQgetvalue(res, i, i_pubviaroot), "t") == 0);
+		pubinfo[i].dacl.acl = pg_strdup(PQgetvalue(res, i, i_pubacl));
+		pubinfo[i].dacl.acldefault = pg_strdup(PQgetvalue(res, i, i_acldefault));
+		pubinfo[i].dacl.privtype = 0;
+		pubinfo[i].dacl.initprivs = NULL;
+		pubinfo[i].dobj.components |= DUMP_COMPONENT_ACL;
 
 		/* Decide whether we want to dump it */
 		selectDumpableObject(&(pubinfo[i].dobj), fout);
@@ -4042,6 +4057,11 @@ dumpPublication(Archive *fout, const PublicationInfo *pubinfo)
 					 NULL, pubinfo->rolname,
 					 pubinfo->dobj.catId, 0, pubinfo->dobj.dumpId);
 
+	if (pubinfo->dobj.dump & DUMP_COMPONENT_ACL)
+		dumpACL(fout, pubinfo->dobj.dumpId, InvalidDumpId, "PUBLICATION",
+				pg_strdup(fmtId(pubinfo->dobj.name)), NULL, NULL,
+				pubinfo->rolname, &pubinfo->dacl);
+
 	destroyPQExpBuffer(delq);
 	destroyPQExpBuffer(query);
 	free(qpubname);
diff --git a/src/bin/pg_dump/pg_dump.h b/src/bin/pg_dump/pg_dump.h
index 1d21c2906f1..15168995baa 100644
--- a/src/bin/pg_dump/pg_dump.h
+++ b/src/bin/pg_dump/pg_dump.h
@@ -613,6 +613,7 @@ typedef struct _policyInfo
 typedef struct _PublicationInfo
 {
 	DumpableObject dobj;
+	DumpableAcl dacl;
 	const char *rolname;
 	bool		puballtables;
 	bool		pubinsert;
diff --git a/src/bin/psql/tab-complete.c b/src/bin/psql/tab-complete.c
index 55af9eb04e4..a6a2ec72ece 100644
--- a/src/bin/psql/tab-complete.c
+++ b/src/bin/psql/tab-complete.c
@@ -3835,6 +3835,7 @@ psql_completion(const char *text, int start, int end)
 											"LARGE OBJECT",
 											"PARAMETER",
 											"PROCEDURE",
+											"PUBLICATION",
 											"ROUTINE",
 											"SCHEMA",
 											"SEQUENCE",
@@ -3872,6 +3873,8 @@ psql_completion(const char *text, int start, int end)
 			COMPLETE_WITH_QUERY(Query_for_list_of_languages);
 		else if (TailMatches("PROCEDURE"))
 			COMPLETE_WITH_VERSIONED_SCHEMA_QUERY(Query_for_list_of_procedures);
+		else if (TailMatches("PUBLICATION"))
+			COMPLETE_WITH_VERSIONED_QUERY(Query_for_list_of_publications);
 		else if (TailMatches("ROUTINE"))
 			COMPLETE_WITH_SCHEMA_QUERY(Query_for_list_of_routines);
 		else if (TailMatches("SCHEMA"))
diff --git a/src/include/catalog/pg_publication.h b/src/include/catalog/pg_publication.h
index 48205ba4293..00287fd3f90 100644
--- a/src/include/catalog/pg_publication.h
+++ b/src/include/catalog/pg_publication.h
@@ -54,6 +54,12 @@ CATALOG(pg_publication,6104,PublicationRelationId)
 
 	/* true if partition changes are published using root schema */
 	bool		pubviaroot;
+
+#ifdef CATALOG_VARLEN			/* variable-length fields start here */
+	/* NOTE: These fields are not present in a relcache entry's rd_rel field. */
+	/* access permissions */
+	aclitem		pubacl[1] BKI_DEFAULT(_null_);
+#endif
 } FormData_pg_publication;
 
 /* ----------------
diff --git a/src/include/nodes/parsenodes.h b/src/include/nodes/parsenodes.h
index 73f635b4553..f49dd92ae6d 100644
--- a/src/include/nodes/parsenodes.h
+++ b/src/include/nodes/parsenodes.h
@@ -87,8 +87,8 @@ typedef uint32 AclMode;			/* a bitmask of privilege bits */
 #define ACL_REFERENCES	(1<<5)
 #define ACL_TRIGGER		(1<<6)
 #define ACL_EXECUTE		(1<<7)	/* for functions */
-#define ACL_USAGE		(1<<8)	/* for languages, namespaces, FDWs, and
-								 * servers */
+#define ACL_USAGE		(1<<8)	/* for languages, namespaces, FDWs, servers
+								 * and publications */
 #define ACL_CREATE		(1<<9)	/* for namespaces and databases */
 #define ACL_CREATE_TEMP (1<<10) /* for databases */
 #define ACL_CONNECT		(1<<11) /* for databases */
diff --git a/src/include/utils/acl.h b/src/include/utils/acl.h
index 48f7d72add5..d8408fe95e1 100644
--- a/src/include/utils/acl.h
+++ b/src/include/utils/acl.h
@@ -168,6 +168,7 @@ typedef struct ArrayType Acl;
 #define ACL_ALL_RIGHTS_SCHEMA		(ACL_USAGE|ACL_CREATE)
 #define ACL_ALL_RIGHTS_TABLESPACE	(ACL_CREATE)
 #define ACL_ALL_RIGHTS_TYPE			(ACL_USAGE)
+#define ACL_ALL_RIGHTS_PUBLICATION	(ACL_USAGE)
 
 /* operation codes for pg_*_aclmask */
 typedef enum
@@ -252,6 +253,8 @@ extern AclMode pg_parameter_acl_aclmask(Oid acl_oid, Oid roleid,
 										AclMode mask, AclMaskHow how);
 extern AclMode pg_proc_aclmask(Oid proc_oid, Oid roleid,
 							   AclMode mask, AclMaskHow how);
+extern AclMode pg_publication_aclmask(Oid pubid, Oid roleid,
+									  AclMode mask, AclMaskHow how);
 extern AclMode pg_language_aclmask(Oid lang_oid, Oid roleid,
 								   AclMode mask, AclMaskHow how);
 extern AclMode pg_largeobject_aclmask_snapshot(Oid lobj_oid, Oid roleid,
@@ -283,6 +286,7 @@ extern AclResult pg_parameter_aclcheck(const char *name, Oid roleid,
 extern AclResult pg_parameter_acl_aclcheck(Oid acl_oid, Oid roleid,
 										   AclMode mode);
 extern AclResult pg_proc_aclcheck(Oid proc_oid, Oid roleid, AclMode mode);
+extern AclResult pg_publication_aclcheck(Oid pub_oid, Oid roleid, AclMode mode);
 extern AclResult pg_language_aclcheck(Oid lang_oid, Oid roleid, AclMode mode);
 extern AclResult pg_largeobject_aclcheck_snapshot(Oid lang_oid, Oid roleid,
 												  AclMode mode, Snapshot snapshot);
-- 
2.31.1

--=-=-=--