From 4004260cf1f3443455b7db89a70a50e38d1663b0 Mon Sep 17 00:00:00 2001
From: Antonin Houska <ah@cybertec.at>
Date: Wed, 18 May 2022 10:58:42 +0200
Subject: [PATCH] Draft implementation of USAGE privilege on PUBLICATION.

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

