Add new predefined role pg_create_subscription.

This role can be granted to non-superusers to allow them to issue
CREATE SUBSCRIPTION. The non-superuser must additionally have CREATE
permissions on the database in which the subscription is to be
created.

Most forms of ALTER SUBSCRIPTION, including ALTER SUBSCRIPTION .. SKIP,
now require only that the role performing the operation own the
subscription, or inherit the privileges of the owner. However, to
use ALTER SUBSCRIPTION ... RENAME or ALTER SUBSCRIPTION ... OWNER TO,
you also need CREATE permission on the database. This is similar to
what we do for schemas. To change the owner of a schema, you must also
have permission to SET ROLE to the new owner, similar to what we do
for other object types.

Non-superusers are required to specify a password for authentication
and the remote side must use the password, similar to what is required
for postgres_fdw and dblink.  A superuser who wants a non-superuser to
own a subscription that does not rely on password authentication may
set the new password_required=false property on that subscription. A
non-superuser may not set password_required=false and may not modify a
subscription that already has password_required=false.

This new password_required subscription property works much like the
eponymous postgres_fdw property.  In both cases, the actual semantics
are that a password is not required if either (1) the property is set
to false or (2) the relevant user is the superuser.

Patch by me, reviewed by Andres Freund, Jeff Davis, Mark Dilger,
and Stephen Frost (but some of those people did not fully endorse
all of the decisions that the patch makes).

Discussion: http://postgr.es/m/CA+TgmoaDH=0Xj7OBiQnsHTKcF2c4L+=gzPBUKSJLh8zed2_+Dg@mail.gmail.com

doc/src/sgml/ref/alter_subscription.sgml

@@ -46,10 +46,11 @@ ALTER SUBSCRIPTION <replaceable class="parameter">name</replaceable> RENAME TO <
 
   <para>
    You must own the subscription to use <command>ALTER SUBSCRIPTION</command>.
-   To alter the owner, you must be able to <literal>SET ROLE</literal> to the
-   new owning role.  The new owner has to be a superuser.
-   (Currently, all subscription owners must be superusers, so the owner checks
-   will be bypassed in practice.  But this might change in the future.)
+   To rename a subscription or alter the owner, you must have
+   <literal>CREATE</literal> permission on the database. In addition,
+   to alter the owner, you must be able to <literal>SET ROLE</literal> to the
+   new owning role. If the subscription has
+   <literal>password_required=false</literal>, only superusers can modify it.
   </para>
 
   <para>
@@ -223,7 +224,9 @@ ALTER SUBSCRIPTION <replaceable class="parameter">name</replaceable> RENAME TO <
       <link linkend="sql-createsubscription-with-binary"><literal>binary</literal></link>,
       <link linkend="sql-createsubscription-with-streaming"><literal>streaming</literal></link>,
       <link linkend="sql-createsubscription-with-disable-on-error"><literal>disable_on_error</literal></link>,
-      and <link linkend="sql-createsubscription-with-origin"><literal>origin</literal></link>.
+      <link linkend="sql-createsubscription-with-password-required"><literal>password_required</literal></link>, and
+      <link linkend="sql-createsubscription-with-origin"><literal>origin</literal></link>.
+      Only a superuser can set <literal>password_required = false</literal>.
      </para>
     </listitem>
    </varlistentry>
@@ -244,8 +247,7 @@ ALTER SUBSCRIPTION <replaceable class="parameter">name</replaceable> RENAME TO <
       finishes a transaction, the LSN (stored in
       <structname>pg_subscription</structname>.<structfield>subskiplsn</structfield>)
       is cleared.  See <xref linkend="logical-replication-conflicts"/> for
-      the details of logical replication conflicts.  Using this command requires
-      superuser privilege.
+      the details of logical replication conflicts.
      </para>
 
      <para>

doc/src/sgml/ref/create_subscription.sgml

@@ -33,7 +33,8 @@ CREATE SUBSCRIPTION <replaceable class="parameter">subscription_name</replaceabl
 
   <para>
    <command>CREATE SUBSCRIPTION</command> adds a new logical-replication
-   subscription.  The subscription name must be distinct from the name of
+   subscription.  The user that creates a subscription becomes the owner
+   of the subscription. The subscription name must be distinct from the name of
    any existing subscription in the current database.
   </para>
 
@@ -49,6 +50,12 @@ CREATE SUBSCRIPTION <replaceable class="parameter">subscription_name</replaceabl
    unless the subscription is initially disabled.
   </para>
 
+  <para>
+   To be able to create a subscription, you must have the privileges of the
+   the <literal>pg_create_subscription</literal> role, as well as
+   <literal>CREATE</literal> privileges on the current database.
+  </para>
+
   <para>
    Additional information about subscriptions and logical replication as a
    whole is available at <xref linkend="logical-replication-subscription"/> and
@@ -365,6 +372,19 @@ CREATE SUBSCRIPTION <replaceable class="parameter">subscription_name</replaceabl
          </para>
         </listitem>
        </varlistentry>
+
+       <varlistentry>
+        <term><literal>password_required</literal> (<type>string</type>)</term>
+        <listitem>
+         <para>
+          Specifies whether connections to the publisher made as a result
+          of this subscription must use password authentication. This setting
+          is ignored when the subscription is owned by a superuser.
+          The default is <literal>true</literal>. Only superusers can set
+          this value to <literal>false</literal>.
+         </para>
+        </listitem>
+       </varlistentry>
       </variablelist></para>
 
     </listitem>

doc/src/sgml/ref/drop_subscription.sgml

@@ -34,7 +34,7 @@ DROP SUBSCRIPTION [ IF EXISTS ] <replaceable class="parameter">name</replaceable
   </para>
 
   <para>
-   A subscription can only be dropped by a superuser.
+   To execute this command the user must be the owner of the subscription.
   </para>
 
   <para>

doc/src/sgml/user-manag.sgml

@@ -699,6 +699,12 @@ DROP ROLE doomed_role;
        <entry>Allow use of connection slots reserved via
        <xref linkend="guc-reserved-connections"/>.</entry>
       </row>
+      <row>
+       <entry>pg_create_subscription</entry>
+       <entry>Allow users with <literal>CREATE</literal> permission on the
+       database to issue
+       <link linkend="sql-createsubscription"><command>CREATE SUBSCRIPTION</command></link>.</entry>
+      </row>
      </tbody>
     </tgroup>
    </table>

src/backend/catalog/pg_subscription.c

@@ -71,6 +71,7 @@ GetSubscription(Oid subid, bool missing_ok)
 	sub->stream = subform->substream;
 	sub->twophasestate = subform->subtwophasestate;
 	sub->disableonerr = subform->subdisableonerr;
+	sub->passwordrequired = subform->subpasswordrequired;
 
 	/* Get conninfo */
 	datum = SysCacheGetAttrNotNull(SUBSCRIPTIONOID,

src/backend/catalog/system_views.sql

@@ -1318,6 +1318,7 @@ REVOKE ALL ON pg_replication_origin_status FROM public;
 REVOKE ALL ON pg_subscription FROM public;
 GRANT SELECT (oid, subdbid, subskiplsn, subname, subowner, subenabled,
               subbinary, substream, subtwophasestate, subdisableonerr,
+			  subpasswordrequired,
               subslotname, subsynccommit, subpublications, suborigin)
     ON pg_subscription TO public;
 

src/backend/commands/alter.c

@@ -24,6 +24,7 @@
 #include "catalog/objectaccess.h"
 #include "catalog/pg_collation.h"
 #include "catalog/pg_conversion.h"
+#include "catalog/pg_database_d.h"
 #include "catalog/pg_event_trigger.h"
 #include "catalog/pg_foreign_data_wrapper.h"
 #include "catalog/pg_foreign_server.h"
@@ -235,6 +236,29 @@ AlterObjectRename_internal(Relation rel, Oid objectId, const char *new_name)
 				aclcheck_error(aclresult, OBJECT_SCHEMA,
 							   get_namespace_name(namespaceId));
 		}
+
+		if (classId == SubscriptionRelationId)
+		{
+			Form_pg_subscription form;
+
+			/* must have CREATE privilege on database */
+			aclresult = object_aclcheck(DatabaseRelationId, MyDatabaseId,
+										GetUserId(), ACL_CREATE);
+			if (aclresult != ACLCHECK_OK)
+				aclcheck_error(aclresult, OBJECT_DATABASE,
+							   get_database_name(MyDatabaseId));
+
+			/*
+			 * Don't allow non-superuser modification of a subscription with
+			 * password_required=false.
+			 */
+			form = (Form_pg_subscription) GETSTRUCT(oldtup);
+			if (!form->subpasswordrequired && !superuser())
+				ereport(ERROR,
+						(errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
+						 errmsg("password_required=false is superuser-only"),
+						 errhint("Subscriptions with the password_required option set to false may only be created or modified by the superuser.")));
+		}
 	}
 
 	/*

src/backend/commands/subscriptioncmds.c

@@ -23,9 +23,12 @@
 #include "catalog/namespace.h"
 #include "catalog/objectaccess.h"
 #include "catalog/objectaddress.h"
+#include "catalog/pg_authid_d.h"
+#include "catalog/pg_database_d.h"
 #include "catalog/pg_subscription.h"
 #include "catalog/pg_subscription_rel.h"
 #include "catalog/pg_type.h"
+#include "commands/dbcommands.h"
 #include "commands/defrem.h"
 #include "commands/event_trigger.h"
 #include "commands/subscriptioncmds.h"
@@ -64,8 +67,9 @@
 #define SUBOPT_STREAMING			0x00000100
 #define SUBOPT_TWOPHASE_COMMIT		0x00000200
 #define SUBOPT_DISABLE_ON_ERR		0x00000400
-#define SUBOPT_LSN					0x00000800
-#define SUBOPT_ORIGIN				0x00001000
+#define SUBOPT_PASSWORD_REQUIRED	0x00000800
+#define SUBOPT_LSN					0x00001000
+#define SUBOPT_ORIGIN				0x00002000
 
 /* check if the 'val' has 'bits' set */
 #define IsSet(val, bits)  (((val) & (bits)) == (bits))
@@ -88,6 +92,7 @@ typedef struct SubOpts
 	char		streaming;
 	bool		twophase;
 	bool		disableonerr;
+	bool		passwordrequired;
 	char	   *origin;
 	XLogRecPtr	lsn;
 } SubOpts;
@@ -144,6 +149,8 @@ parse_subscription_options(ParseState *pstate, List *stmt_options,
 		opts->twophase = false;
 	if (IsSet(supported_opts, SUBOPT_DISABLE_ON_ERR))
 		opts->disableonerr = false;
+	if (IsSet(supported_opts, SUBOPT_PASSWORD_REQUIRED))
+		opts->passwordrequired = true;
 	if (IsSet(supported_opts, SUBOPT_ORIGIN))
 		opts->origin = pstrdup(LOGICALREP_ORIGIN_ANY);
 
@@ -274,6 +281,15 @@ parse_subscription_options(ParseState *pstate, List *stmt_options,
 			opts->specified_opts |= SUBOPT_DISABLE_ON_ERR;
 			opts->disableonerr = defGetBoolean(defel);
 		}
+		else if (IsSet(supported_opts, SUBOPT_PASSWORD_REQUIRED) &&
+				 strcmp(defel->defname, "password_required") == 0)
+		{
+			if (IsSet(opts->specified_opts, SUBOPT_PASSWORD_REQUIRED))
+				errorConflictingDefElem(defel, pstate);
+
+			opts->specified_opts |= SUBOPT_PASSWORD_REQUIRED;
+			opts->passwordrequired = defGetBoolean(defel);
+		}
 		else if (IsSet(supported_opts, SUBOPT_ORIGIN) &&
 				 strcmp(defel->defname, "origin") == 0)
 		{
@@ -550,6 +566,7 @@ CreateSubscription(ParseState *pstate, CreateSubscriptionStmt *stmt,
 	List	   *publications;
 	bits32		supported_opts;
 	SubOpts		opts = {0};
+	AclResult	aclresult;
 
 	/*
 	 * Parse and check options.
@@ -560,7 +577,8 @@ CreateSubscription(ParseState *pstate, CreateSubscriptionStmt *stmt,
 					  SUBOPT_SLOT_NAME | SUBOPT_COPY_DATA |
 					  SUBOPT_SYNCHRONOUS_COMMIT | SUBOPT_BINARY |
 					  SUBOPT_STREAMING | SUBOPT_TWOPHASE_COMMIT |
-					  SUBOPT_DISABLE_ON_ERR | SUBOPT_ORIGIN);
+					  SUBOPT_DISABLE_ON_ERR | SUBOPT_PASSWORD_REQUIRED |
+					  SUBOPT_ORIGIN);
 	parse_subscription_options(pstate, stmt->options, supported_opts, &opts);
 
 	/*
@@ -572,10 +590,36 @@ CreateSubscription(ParseState *pstate, CreateSubscriptionStmt *stmt,
 	if (opts.create_slot)
 		PreventInTransactionBlock(isTopLevel, "CREATE SUBSCRIPTION ... WITH (create_slot = true)");
 
-	if (!superuser())
+	/*
+	 * We don't want to allow unprivileged users to be able to trigger attempts
+	 * to access arbitrary network destinations, so require the user to have
+	 * been specifically authorized to create subscriptions.
+	 */
+	if (!has_privs_of_role(owner, ROLE_PG_CREATE_SUBSCRIPTION))
 		ereport(ERROR,
 				(errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
-				 errmsg("must be superuser to create subscriptions")));
+				 errmsg("must have privileges of pg_create_subscription to create subscriptions")));
+
+	/*
+	 * Since a subscription is a database object, we also check for CREATE
+	 * permission on the database.
+	 */
+	aclresult = object_aclcheck(DatabaseRelationId, MyDatabaseId,
+								owner, ACL_CREATE);
+	if (aclresult != ACLCHECK_OK)
+		aclcheck_error(aclresult, OBJECT_DATABASE,
+					   get_database_name(MyDatabaseId));
+
+	/*
+	 * Non-superusers are required to set a password for authentication, and
+	 * that password must be used by the target server, but the superuser can
+	 * exempt a subscription from this requirement.
+	 */
+	if (!opts.passwordrequired && !superuser_arg(owner))
+			ereport(ERROR,
+					(errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
+					 errmsg("password_required=false is superuser-only"),
+					 errhint("Subscriptions with the password_required option set to false may only be created or modified by the superuser.")));
 
 	/*
 	 * If built with appropriate switch, whine when regression-testing
@@ -614,7 +658,7 @@ CreateSubscription(ParseState *pstate, CreateSubscriptionStmt *stmt,
 	load_file("libpqwalreceiver", false);
 
 	/* Check the connection info string. */
-	walrcv_check_conninfo(conninfo);
+	walrcv_check_conninfo(conninfo, opts.passwordrequired && !superuser());
 
 	/* Everything ok, form a new tuple. */
 	memset(values, 0, sizeof(values));
@@ -636,6 +680,7 @@ CreateSubscription(ParseState *pstate, CreateSubscriptionStmt *stmt,
 					 LOGICALREP_TWOPHASE_STATE_PENDING :
 					 LOGICALREP_TWOPHASE_STATE_DISABLED);
 	values[Anum_pg_subscription_subdisableonerr - 1] = BoolGetDatum(opts.disableonerr);
+	values[Anum_pg_subscription_subpasswordrequired - 1] = BoolGetDatum(opts.passwordrequired);
 	values[Anum_pg_subscription_subconninfo - 1] =
 		CStringGetTextDatum(conninfo);
 	if (opts.slot_name)
@@ -672,9 +717,12 @@ CreateSubscription(ParseState *pstate, CreateSubscriptionStmt *stmt,
 		List	   *tables;
 		ListCell   *lc;
 		char		table_state;
+		bool		must_use_password;
 
 		/* Try to connect to the publisher. */
-		wrconn = walrcv_connect(conninfo, true, stmt->subname, &err);
+		must_use_password = !superuser_arg(owner) && opts.passwordrequired;
+		wrconn = walrcv_connect(conninfo, true, must_use_password,
+								stmt->subname, &err);
 		if (!wrconn)
 			ereport(ERROR,
 					(errcode(ERRCODE_CONNECTION_FAILURE),
@@ -799,12 +847,15 @@ AlterSubscription_refresh(Subscription *sub, bool copy_data,
 	} SubRemoveRels;
 	SubRemoveRels *sub_remove_rels;
 	WalReceiverConn *wrconn;
+	bool		must_use_password;
 
 	/* Load the library providing us libpq calls. */
 	load_file("libpqwalreceiver", false);
 
 	/* Try to connect to the publisher. */
-	wrconn = walrcv_connect(sub->conninfo, true, sub->name, &err);
+	must_use_password = !superuser_arg(sub->owner) && sub->passwordrequired;
+	wrconn = walrcv_connect(sub->conninfo, true, must_use_password,
+							sub->name, &err);
 	if (!wrconn)
 		ereport(ERROR,
 				(errcode(ERRCODE_CONNECTION_FAILURE),
@@ -1039,6 +1090,16 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
 
 	sub = GetSubscription(subid, false);
 
+	/*
+	 * Don't allow non-superuser modification of a subscription with
+	 * password_required=false.
+	 */
+	if (!sub->passwordrequired && !superuser())
+		ereport(ERROR,
+				(errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
+						 errmsg("password_required=false is superuser-only"),
+						 errhint("Subscriptions with the password_required option set to false may only be created or modified by the superuser.")));
+
 	/* Lock the subscription so nobody else can do anything with it. */
 	LockSharedObject(SubscriptionRelationId, subid, 0, AccessExclusiveLock);
 
@@ -1054,7 +1115,7 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
 				supported_opts = (SUBOPT_SLOT_NAME |
 								  SUBOPT_SYNCHRONOUS_COMMIT | SUBOPT_BINARY |
 								  SUBOPT_STREAMING | SUBOPT_DISABLE_ON_ERR |
-								  SUBOPT_ORIGIN);
+								  SUBOPT_PASSWORD_REQUIRED | SUBOPT_ORIGIN);
 
 				parse_subscription_options(pstate, stmt->options,
 										   supported_opts, &opts);
@@ -1111,6 +1172,21 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
 						= true;
 				}
 
+				if (IsSet(opts.specified_opts, SUBOPT_PASSWORD_REQUIRED))
+				{
+					/* Non-superuser may not disable password_required. */
+					if (!opts.passwordrequired && !superuser())
+						ereport(ERROR,
+								(errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
+								 errmsg("password_required=false is superuser-only"),
+								 errhint("Subscriptions with the password_required option set to false may only be created or modified by the superuser.")));
+
+					values[Anum_pg_subscription_subpasswordrequired - 1]
+						= BoolGetDatum(opts.passwordrequired);
+					replaces[Anum_pg_subscription_subpasswordrequired - 1]
+						= true;
+				}
+
 				if (IsSet(opts.specified_opts, SUBOPT_ORIGIN))
 				{
 					values[Anum_pg_subscription_suborigin - 1] =
@@ -1148,7 +1224,8 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
 			/* Load the library providing us libpq calls. */
 			load_file("libpqwalreceiver", false);
 			/* Check the connection info string. */
-			walrcv_check_conninfo(stmt->conninfo);
+			walrcv_check_conninfo(stmt->conninfo,
+								  sub->passwordrequired && !superuser_arg(sub->owner));
 
 			values[Anum_pg_subscription_subconninfo - 1] =
 				CStringGetTextDatum(stmt->conninfo);
@@ -1305,11 +1382,6 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
 				/* ALTER SUBSCRIPTION ... SKIP supports only LSN option */
 				Assert(IsSet(opts.specified_opts, SUBOPT_LSN));
 
-				if (!superuser())
-					ereport(ERROR,
-							(errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
-							 errmsg("must be superuser to skip transaction")));
-
 				/*
 				 * If the user sets subskiplsn, we do a sanity check to make
 				 * sure that the specified LSN is a probable value.
@@ -1379,6 +1451,7 @@ DropSubscription(DropSubscriptionStmt *stmt, bool isTopLevel)
 	ObjectAddress myself;
 	HeapTuple	tup;
 	Oid			subid;
+	Oid			subowner;
 	Datum		datum;
 	bool		isnull;
 	char	   *subname;
@@ -1391,6 +1464,7 @@ DropSubscription(DropSubscriptionStmt *stmt, bool isTopLevel)
 	WalReceiverConn *wrconn;
 	Form_pg_subscription form;
 	List	   *rstates;
+	bool		must_use_password;
 
 	/*
 	 * Lock pg_subscription with AccessExclusiveLock to ensure that the
@@ -1420,6 +1494,8 @@ DropSubscription(DropSubscriptionStmt *stmt, bool isTopLevel)
 
 	form = (Form_pg_subscription) GETSTRUCT(tup);
 	subid = form->oid;
+	subowner = form->subowner;
+	must_use_password = !superuser_arg(subowner) && form->subpasswordrequired;
 
 	/* must be owner */
 	if (!object_ownercheck(SubscriptionRelationId, subid, GetUserId()))
@@ -1576,7 +1652,8 @@ DropSubscription(DropSubscriptionStmt *stmt, bool isTopLevel)
 	 */
 	load_file("libpqwalreceiver", false);
 
-	wrconn = walrcv_connect(conninfo, true, subname, &err);
+	wrconn = walrcv_connect(conninfo, true, must_use_password,
+							subname, &err);
 	if (wrconn == NULL)
 	{
 		if (!slotname)
@@ -1715,6 +1792,7 @@ static void
 AlterSubscriptionOwner_internal(Relation rel, HeapTuple tup, Oid newOwnerId)
 {
 	Form_pg_subscription form;
+	AclResult	aclresult;
 
 	form = (Form_pg_subscription) GETSTRUCT(tup);
 
@@ -1725,13 +1803,31 @@ AlterSubscriptionOwner_internal(Relation rel, HeapTuple tup, Oid newOwnerId)
 		aclcheck_error(ACLCHECK_NOT_OWNER, OBJECT_SUBSCRIPTION,
 					   NameStr(form->subname));
 
-	/* New owner must be a superuser */
-	if (!superuser_arg(newOwnerId))
+	/*
+	 * Don't allow non-superuser modification of a subscription with
+	 * password_required=false.
+	 */
+	if (!form->subpasswordrequired && !superuser())
 		ereport(ERROR,
 				(errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
-				 errmsg("permission denied to change owner of subscription \"%s\"",
-						NameStr(form->subname)),
-				 errhint("The owner of a subscription must be a superuser.")));
+						 errmsg("password_required=false is superuser-only"),
+						 errhint("Subscriptions with the password_required option set to false may only be created or modified by the superuser.")));
+
+	/* Must be able to become new owner */
+	check_can_set_role(GetUserId(), newOwnerId);
+
+	/*
+	 * current owner must have CREATE on database
+	 *
+	 * This is consistent with how ALTER SCHEMA ... OWNER TO works, but some
+	 * other object types behave differently (e.g. you can't give a table to
+	 * a user who lacks CREATE privileges on a schema).
+	 */
+	aclresult = object_aclcheck(DatabaseRelationId, MyDatabaseId,
+								GetUserId(), ACL_CREATE);
+	if (aclresult != ACLCHECK_OK)
+		aclcheck_error(aclresult, OBJECT_DATABASE,
+					   get_database_name(MyDatabaseId));
 
 	form->subowner = newOwnerId;
 	CatalogTupleUpdate(rel, &tup->t_self, tup);

src/backend/replication/libpqwalreceiver/libpqwalreceiver.c

@@ -49,9 +49,10 @@ struct WalReceiverConn
 
 /* Prototypes for interface functions */
 static WalReceiverConn *libpqrcv_connect(const char *conninfo,
-										 bool logical, const char *appname,
-										 char **err);
-static void libpqrcv_check_conninfo(const char *conninfo);
+										 bool logical, bool must_use_password,
+										 const char *appname, char **err);
+static void libpqrcv_check_conninfo(const char *conninfo,
+									bool must_use_password);
 static char *libpqrcv_get_conninfo(WalReceiverConn *conn);
 static void libpqrcv_get_senderinfo(WalReceiverConn *conn,
 									char **sender_host, int *sender_port);
@@ -119,11 +120,17 @@ _PG_init(void)
 /*
  * Establish the connection to the primary server for XLOG streaming
  *
- * Returns NULL on error and fills the err with palloc'ed error message.
+ * If an error occurs, this function will normally return NULL and set *err
+ * to a palloc'ed error message. However, if must_use_password is true and
+ * the connection fails to use the password, this function will ereport(ERROR).
+ * We do this because in that case the error includes a detail and a hint for
+ * consistency with other parts of the system, and it's not worth adding the
+ * machinery to pass all of those back to the caller just to cover this one
+ * case.
  */
 static WalReceiverConn *
-libpqrcv_connect(const char *conninfo, bool logical, const char *appname,
-				 char **err)
+libpqrcv_connect(const char *conninfo, bool logical, bool must_use_password,
+				 const char *appname, char **err)
 {
 	WalReceiverConn *conn;
 	const char *keys[6];
@@ -180,6 +187,18 @@ libpqrcv_connect(const char *conninfo, bool logical, const char *appname,
 	if (PQstatus(conn->streamConn) != CONNECTION_OK)
 		goto bad_connection_errmsg;
 
+	if (must_use_password && !PQconnectionUsedPassword(conn->streamConn))
+	{
+		libpqsrv_disconnect(conn->streamConn);
+		pfree(conn);
+
+		ereport(ERROR,
+				(errcode(ERRCODE_S_R_E_PROHIBITED_SQL_STATEMENT_ATTEMPTED),
+				 errmsg("password is required"),
+				 errdetail("Non-superuser cannot connect if the server does not request a password."),
+				 errhint("Target server's authentication method must be changed. or set password_required=false in the subscription attributes.")));
+	}
+
 	if (logical)
 	{
 		PGresult   *res;
@@ -212,12 +231,18 @@ bad_connection:
 }
 
 /*
- * Validate connection info string (just try to parse it)
+ * Validate connection info string, and determine whether it might cause
+ * local filesystem access to be attempted.
+ *
+ * If the connection string can't be parsed, this function will raise
+ * an error and will not return. If it can, it will return true if this
+ * connection string specifies a password and false otherwise.
  */
 static void
-libpqrcv_check_conninfo(const char *conninfo)
+libpqrcv_check_conninfo(const char *conninfo, bool must_use_password)
 {
 	PQconninfoOption *opts = NULL;
+	PQconninfoOption *opt;
 	char	   *err = NULL;
 
 	opts = PQconninfoParse(conninfo, &err);
@@ -232,6 +257,30 @@ libpqrcv_check_conninfo(const char *conninfo)
 				 errmsg("invalid connection string syntax: %s", errcopy)));
 	}
 
+	if (must_use_password)
+	{
+		bool	uses_password = false;
+
+		for (opt = opts; opt->keyword != NULL; ++opt)
+		{
+			/* Ignore connection options that are not present. */
+			if (opt->val == NULL)
+				continue;
+
+			if (strcmp(opt->keyword, "password") == 0 && opt->val[0] != '\0')
+			{
+				uses_password = true;
+				break;
+			}
+		}
+
+		if (!uses_password)
+			ereport(ERROR,
+					(errcode(ERRCODE_S_R_E_PROHIBITED_SQL_STATEMENT_ATTEMPTED),
+					 errmsg("password is required"),
+					 errdetail("Non-superusers must provide a password in the connection string.")));
+	}
+
 	PQconninfoFree(opts);
 }
 

src/backend/replication/logical/tablesync.c

@@ -1252,6 +1252,7 @@ LogicalRepSyncTableStart(XLogRecPtr *origin_startpos)
 	WalRcvExecResult *res;
 	char		originname[NAMEDATALEN];
 	RepOriginId originid;
+	bool		must_use_password;
 
 	/* Check the state of the table synchronization. */
 	StartTransactionCommand();
@@ -1284,13 +1285,19 @@ LogicalRepSyncTableStart(XLogRecPtr *origin_startpos)
 									slotname,
 									NAMEDATALEN);
 
+	/* Is the use of a password mandatory? */
+	must_use_password = MySubscription->passwordrequired &&
+		!superuser_arg(MySubscription->owner);
+
 	/*
 	 * Here we use the slot name instead of the subscription name as the
 	 * application_name, so that it is different from the leader apply worker,
 	 * so that synchronous replication can distinguish them.
 	 */
 	LogRepWorkerWalRcvConn =
-		walrcv_connect(MySubscription->conninfo, true, slotname, &err);
+		walrcv_connect(MySubscription->conninfo, true,
+					   must_use_password,
+					   slotname, &err);
 	if (LogRepWorkerWalRcvConn == NULL)
 		ereport(ERROR,
 				(errcode(ERRCODE_CONNECTION_FAILURE),

src/backend/replication/logical/worker.c

@@ -4521,6 +4521,7 @@ ApplyWorkerMain(Datum main_arg)
 		RepOriginId originid;
 		TimeLineID	startpointTLI;
 		char	   *err;
+		bool		must_use_password;
 
 		myslotname = MySubscription->slotname;
 
@@ -4546,7 +4547,12 @@ ApplyWorkerMain(Datum main_arg)
 		origin_startpos = replorigin_session_get_progress(false);
 		CommitTransactionCommand();
 
+		/* Is the use of a password mandatory? */
+		must_use_password = MySubscription->passwordrequired &&
+			!superuser_arg(MySubscription->owner);
+
 		LogRepWorkerWalRcvConn = walrcv_connect(MySubscription->conninfo, true,
+												must_use_password,
 												MySubscription->name, &err);
 		if (LogRepWorkerWalRcvConn == NULL)
 			ereport(ERROR,

src/backend/replication/walreceiver.c

@@ -296,7 +296,7 @@ WalReceiverMain(void)
 	sigprocmask(SIG_SETMASK, &UnBlockSig, NULL);
 
 	/* Establish the connection to the primary for XLOG streaming */
-	wrconn = walrcv_connect(conninfo, false,
+	wrconn = walrcv_connect(conninfo, false, false,
 							cluster_name[0] ? cluster_name : "walreceiver",
 							&err);
 	if (!wrconn)

src/bin/pg_dump/pg_dump.c

@@ -4608,6 +4608,7 @@ getSubscriptions(Archive *fout)
 	int			i_subsynccommit;
 	int			i_subpublications;
 	int			i_subbinary;
+	int			i_subpasswordrequired;
 	int			i,
 				ntups;
 
@@ -4660,9 +4661,14 @@ getSubscriptions(Archive *fout)
 						  LOGICALREP_TWOPHASE_STATE_DISABLED);
 
 	if (fout->remoteVersion >= 160000)
-		appendPQExpBufferStr(query, " s.suborigin\n");
+		appendPQExpBufferStr(query,
+							 " s.suborigin,\n"
+							 " s.subpasswordrequired\n");
 	else
-		appendPQExpBuffer(query, " '%s' AS suborigin\n", LOGICALREP_ORIGIN_ANY);
+		appendPQExpBuffer(query,
+						  " '%s' AS suborigin,\n"
+						  " 't' AS subpasswordrequired\n",
+						  LOGICALREP_ORIGIN_ANY);
 
 	appendPQExpBufferStr(query,
 						 "FROM pg_subscription s\n"
@@ -4690,6 +4696,7 @@ getSubscriptions(Archive *fout)
 	i_subtwophasestate = PQfnumber(res, "subtwophasestate");
 	i_subdisableonerr = PQfnumber(res, "subdisableonerr");
 	i_suborigin = PQfnumber(res, "suborigin");
+	i_subpasswordrequired = PQfnumber(res, "subpasswordrequired");
 
 	subinfo = pg_malloc(ntups * sizeof(SubscriptionInfo));
 
@@ -4720,6 +4727,8 @@ getSubscriptions(Archive *fout)
 		subinfo[i].subdisableonerr =
 			pg_strdup(PQgetvalue(res, i, i_subdisableonerr));
 		subinfo[i].suborigin = pg_strdup(PQgetvalue(res, i, i_suborigin));
+		subinfo[i].subpasswordrequired =
+			pg_strdup(PQgetvalue(res, i, i_subpasswordrequired));
 
 		/* Decide whether we want to dump it */
 		selectDumpableObject(&(subinfo[i].dobj), fout);
@@ -4801,6 +4810,9 @@ dumpSubscription(Archive *fout, const SubscriptionInfo *subinfo)
 	if (strcmp(subinfo->subsynccommit, "off") != 0)
 		appendPQExpBuffer(query, ", synchronous_commit = %s", fmtId(subinfo->subsynccommit));
 
+	if (strcmp(subinfo->subpasswordrequired, "t") != 0)
+		appendPQExpBuffer(query, ", password_required = false");
+
 	appendPQExpBufferStr(query, ");\n");
 
 	if (subinfo->dobj.dump & DUMP_COMPONENT_DEFINITION)

src/bin/pg_dump/pg_dump.h

@@ -663,6 +663,7 @@ typedef struct _SubscriptionInfo
 	char	   *suborigin;
 	char	   *subsynccommit;
 	char	   *subpublications;
+	char	   *subpasswordrequired;
 } SubscriptionInfo;
 
 /*

src/include/catalog/catversion.h

@@ -57,6 +57,6 @@
  */
 
 /*							yyyymmddN */
-#define CATALOG_VERSION_NO	202303293
+#define CATALOG_VERSION_NO	202303301
 
 #endif

src/include/catalog/pg_authid.dat

@@ -94,5 +94,10 @@
   rolcreaterole => 'f', rolcreatedb => 'f', rolcanlogin => 'f',
   rolreplication => 'f', rolbypassrls => 'f', rolconnlimit => '-1',
   rolpassword => '_null_', rolvaliduntil => '_null_' },
+{ oid => '9535', oid_symbol => 'ROLE_PG_CREATE_SUBSCRIPTION',
+  rolname => 'pg_create_subscription', rolsuper => 'f', rolinherit => 't',
+  rolcreaterole => 'f', rolcreatedb => 'f', rolcanlogin => 'f',
+  rolreplication => 'f', rolbypassrls => 'f', rolconnlimit => '-1',
+  rolpassword => '_null_', rolvaliduntil => '_null_' },
 
 ]

src/include/catalog/pg_subscription.h

@@ -88,6 +88,8 @@ CATALOG(pg_subscription,6100,SubscriptionRelationId) BKI_SHARED_RELATION BKI_ROW
 	bool		subdisableonerr;	/* True if a worker error should cause the
 									 * subscription to be disabled */
 
+	bool		subpasswordrequired; /* Must connection use a password? */
+
 #ifdef CATALOG_VARLEN			/* variable-length fields start here */
 	/* Connection string to the publisher */
 	text		subconninfo BKI_FORCE_NOT_NULL;
@@ -131,6 +133,7 @@ typedef struct Subscription
 	bool		disableonerr;	/* Indicates if the subscription should be
 								 * automatically disabled if a worker error
 								 * occurs */
+	bool		passwordrequired;	/* Must connection use a password? */
 	char	   *conninfo;		/* Connection string to the publisher */
 	char	   *slotname;		/* Name of the replication slot */
 	char	   *synccommit;		/* Synchronous commit setting for worker */

src/include/replication/walreceiver.h

@@ -239,6 +239,7 @@ typedef struct WalRcvExecResult
  */
 typedef WalReceiverConn *(*walrcv_connect_fn) (const char *conninfo,
 											   bool logical,
+											   bool must_use_password,
 											   const char *appname,
 											   char **err);
 
@@ -247,7 +248,8 @@ typedef WalReceiverConn *(*walrcv_connect_fn) (const char *conninfo,
  *
  * Parse and validate the connection string given as of 'conninfo'.
  */
-typedef void (*walrcv_check_conninfo_fn) (const char *conninfo);
+typedef void (*walrcv_check_conninfo_fn) (const char *conninfo,
+										  bool must_use_password);
 
 /*
  * walrcv_get_conninfo_fn
@@ -405,10 +407,10 @@ typedef struct WalReceiverFunctionsType
 
 extern PGDLLIMPORT WalReceiverFunctionsType *WalReceiverFunctions;
 
-#define walrcv_connect(conninfo, logical, appname, err) \
-	WalReceiverFunctions->walrcv_connect(conninfo, logical, appname, err)
-#define walrcv_check_conninfo(conninfo) \
-	WalReceiverFunctions->walrcv_check_conninfo(conninfo)
+#define walrcv_connect(conninfo, logical, must_use_password, appname, err) \
+	WalReceiverFunctions->walrcv_connect(conninfo, logical, must_use_password, appname, err)
+#define walrcv_check_conninfo(conninfo, must_use_password) \
+	WalReceiverFunctions->walrcv_check_conninfo(conninfo, must_use_password)
 #define walrcv_get_conninfo(conn) \
 	WalReceiverFunctions->walrcv_get_conninfo(conn)
 #define walrcv_get_senderinfo(conn, sender_host, sender_port) \

src/test/regress/expected/subscription.out

@@ -3,6 +3,7 @@
 --
 CREATE ROLE regress_subscription_user LOGIN SUPERUSER;
 CREATE ROLE regress_subscription_user2;
+CREATE ROLE regress_subscription_user3 IN ROLE pg_create_subscription;
 CREATE ROLE regress_subscription_user_dummy LOGIN NOSUPERUSER;
 SET SESSION AUTHORIZATION 'regress_subscription_user';
 -- fail - no publications
@@ -78,7 +79,7 @@ ERROR:  subscription "regress_testsub" already exists
 -- fail - must be superuser
 SET SESSION AUTHORIZATION 'regress_subscription_user2';
 CREATE SUBSCRIPTION regress_testsub2 CONNECTION 'dbname=regress_doesnotexist' PUBLICATION foo WITH (connect = false);
-ERROR:  must be superuser to create subscriptions
+ERROR:  must have privileges of pg_create_subscription to create subscriptions
 SET SESSION AUTHORIZATION 'regress_subscription_user';
 -- fail - invalid option combinations
 CREATE SUBSCRIPTION regress_testsub2 CONNECTION 'dbname=regress_doesnotexist' PUBLICATION testpub WITH (connect = false, copy_data = true);
@@ -152,6 +153,8 @@ ERROR:  invalid connection string syntax: missing "=" after "foobar" in connecti
 ALTER SUBSCRIPTION regress_testsub SET PUBLICATION testpub2, testpub3 WITH (refresh = false);
 ALTER SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist2';
 ALTER SUBSCRIPTION regress_testsub SET (slot_name = 'newname');
+ALTER SUBSCRIPTION regress_testsub SET (password_required = false);
+ALTER SUBSCRIPTION regress_testsub SET (password_required = true);
 -- fail
 ALTER SUBSCRIPTION regress_testsub SET (slot_name = '');
 ERROR:  replication slot name "" is too short
@@ -218,12 +221,7 @@ HINT:  Available values: local, remote_write, remote_apply, on, off.
 
 -- rename back to keep the rest simple
 ALTER SUBSCRIPTION regress_testsub_foo RENAME TO regress_testsub;
--- fail - new owner must be superuser
-ALTER SUBSCRIPTION regress_testsub OWNER TO regress_subscription_user2;
-ERROR:  permission denied to change owner of subscription "regress_testsub"
-HINT:  The owner of a subscription must be a superuser.
-ALTER ROLE regress_subscription_user2 SUPERUSER;
--- now it works
+-- ok, we're a superuser
 ALTER SUBSCRIPTION regress_testsub OWNER TO regress_subscription_user2;
 -- fail - cannot do DROP SUBSCRIPTION inside transaction block with slot name
 BEGIN;
@@ -418,6 +416,51 @@ ALTER SUBSCRIPTION regress_testsub SET (disable_on_error = true);
  regress_testsub | regress_subscription_user | f       | {testpub}   | f      | off       | d                | t                | any    | off                | dbname=regress_doesnotexist | 0/0
 (1 row)
 
+ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE);
+DROP SUBSCRIPTION regress_testsub;
+-- let's do some tests with pg_create_subscription rather than superuser
+SET SESSION AUTHORIZATION regress_subscription_user3;
+-- fail, not enough privileges
+CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUBLICATION testpub WITH (connect = false);
+ERROR:  permission denied for database regression
+-- fail, must specify password
+RESET SESSION AUTHORIZATION;
+GRANT CREATE ON DATABASE REGRESSION TO regress_subscription_user3;
+SET SESSION AUTHORIZATION regress_subscription_user3;
+CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUBLICATION testpub WITH (connect = false);
+ERROR:  password is required
+DETAIL:  Non-superusers must provide a password in the connection string.
+-- fail, can't set password_required=false
+RESET SESSION AUTHORIZATION;
+GRANT CREATE ON DATABASE REGRESSION TO regress_subscription_user3;
+SET SESSION AUTHORIZATION regress_subscription_user3;
+CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUBLICATION testpub WITH (connect = false, password_required = false);
+ERROR:  password_required=false is superuser-only
+HINT:  Subscriptions with the password_required option set to false may only be created or modified by the superuser.
+-- ok
+RESET SESSION AUTHORIZATION;
+GRANT CREATE ON DATABASE REGRESSION TO regress_subscription_user3;
+SET SESSION AUTHORIZATION regress_subscription_user3;
+CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist password=regress_fakepassword' PUBLICATION testpub WITH (connect = false);
+WARNING:  subscription was created, but is not connected
+HINT:  To initiate replication, you must manually create the replication slot, enable the subscription, and refresh the subscription.
+-- we cannot give the subscription away to some random user
+ALTER SUBSCRIPTION regress_testsub OWNER TO regress_subscription_user;
+ERROR:  must be able to SET ROLE "regress_subscription_user"
+-- but we can rename the subscription we just created
+ALTER SUBSCRIPTION regress_testsub RENAME TO regress_testsub2;
+-- ok, even after losing pg_create_subscription we can still rename it
+RESET SESSION AUTHORIZATION;
+REVOKE pg_create_subscription FROM regress_subscription_user3;
+SET SESSION AUTHORIZATION regress_subscription_user3;
+ALTER SUBSCRIPTION regress_testsub2 RENAME TO regress_testsub;
+-- fail, after losing CREATE on the database we can't rename it any more
+RESET SESSION AUTHORIZATION;
+REVOKE CREATE ON DATABASE REGRESSION FROM regress_subscription_user3;
+SET SESSION AUTHORIZATION regress_subscription_user3;
+ALTER SUBSCRIPTION regress_testsub RENAME TO regress_testsub2;
+ERROR:  permission denied for database regression
+-- ok, owning it is enough for this stuff
 ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE);
 DROP SUBSCRIPTION regress_testsub;
 RESET SESSION AUTHORIZATION;

src/test/regress/sql/subscription.sql

@@ -4,6 +4,7 @@
 
 CREATE ROLE regress_subscription_user LOGIN SUPERUSER;
 CREATE ROLE regress_subscription_user2;
+CREATE ROLE regress_subscription_user3 IN ROLE pg_create_subscription;
 CREATE ROLE regress_subscription_user_dummy LOGIN NOSUPERUSER;
 SET SESSION AUTHORIZATION 'regress_subscription_user';
 
@@ -92,6 +93,8 @@ ALTER SUBSCRIPTION regress_testsub CONNECTION 'foobar';
 ALTER SUBSCRIPTION regress_testsub SET PUBLICATION testpub2, testpub3 WITH (refresh = false);
 ALTER SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist2';
 ALTER SUBSCRIPTION regress_testsub SET (slot_name = 'newname');
+ALTER SUBSCRIPTION regress_testsub SET (password_required = false);
+ALTER SUBSCRIPTION regress_testsub SET (password_required = true);
 
 -- fail
 ALTER SUBSCRIPTION regress_testsub SET (slot_name = '');
@@ -138,10 +141,7 @@ ALTER SUBSCRIPTION regress_testsub_foo SET (synchronous_commit = foobar);
 -- rename back to keep the rest simple
 ALTER SUBSCRIPTION regress_testsub_foo RENAME TO regress_testsub;
 
--- fail - new owner must be superuser
-ALTER SUBSCRIPTION regress_testsub OWNER TO regress_subscription_user2;
-ALTER ROLE regress_subscription_user2 SUPERUSER;
--- now it works
+-- ok, we're a superuser
 ALTER SUBSCRIPTION regress_testsub OWNER TO regress_subscription_user2;
 
 -- fail - cannot do DROP SUBSCRIPTION inside transaction block with slot name
@@ -286,6 +286,52 @@ ALTER SUBSCRIPTION regress_testsub SET (disable_on_error = true);
 ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE);
 DROP SUBSCRIPTION regress_testsub;
 
+-- let's do some tests with pg_create_subscription rather than superuser
+SET SESSION AUTHORIZATION regress_subscription_user3;
+
+-- fail, not enough privileges
+CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUBLICATION testpub WITH (connect = false);
+
+-- fail, must specify password
+RESET SESSION AUTHORIZATION;
+GRANT CREATE ON DATABASE REGRESSION TO regress_subscription_user3;
+SET SESSION AUTHORIZATION regress_subscription_user3;
+CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUBLICATION testpub WITH (connect = false);
+
+-- fail, can't set password_required=false
+RESET SESSION AUTHORIZATION;
+GRANT CREATE ON DATABASE REGRESSION TO regress_subscription_user3;
+SET SESSION AUTHORIZATION regress_subscription_user3;
+CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUBLICATION testpub WITH (connect = false, password_required = false);
+
+-- ok
+RESET SESSION AUTHORIZATION;
+GRANT CREATE ON DATABASE REGRESSION TO regress_subscription_user3;
+SET SESSION AUTHORIZATION regress_subscription_user3;
+CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist password=regress_fakepassword' PUBLICATION testpub WITH (connect = false);
+
+-- we cannot give the subscription away to some random user
+ALTER SUBSCRIPTION regress_testsub OWNER TO regress_subscription_user;
+
+-- but we can rename the subscription we just created
+ALTER SUBSCRIPTION regress_testsub RENAME TO regress_testsub2;
+
+-- ok, even after losing pg_create_subscription we can still rename it
+RESET SESSION AUTHORIZATION;
+REVOKE pg_create_subscription FROM regress_subscription_user3;
+SET SESSION AUTHORIZATION regress_subscription_user3;
+ALTER SUBSCRIPTION regress_testsub2 RENAME TO regress_testsub;
+
+-- fail, after losing CREATE on the database we can't rename it any more
+RESET SESSION AUTHORIZATION;
+REVOKE CREATE ON DATABASE REGRESSION FROM regress_subscription_user3;
+SET SESSION AUTHORIZATION regress_subscription_user3;
+ALTER SUBSCRIPTION regress_testsub RENAME TO regress_testsub2;
+
+-- ok, owning it is enough for this stuff
+ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE);
+DROP SUBSCRIPTION regress_testsub;
+
 RESET SESSION AUTHORIZATION;
 DROP ROLE regress_subscription_user;
 DROP ROLE regress_subscription_user2;

src/test/subscription/t/027_nosuperuser.pl

@@ -150,7 +150,7 @@ CREATE PUBLICATION alice
 $node_subscriber->safe_psql(
 	'postgres', qq(
 SET SESSION AUTHORIZATION regress_admin;
-CREATE SUBSCRIPTION admin_sub CONNECTION '$publisher_connstr' PUBLICATION alice;
+CREATE SUBSCRIPTION admin_sub CONNECTION '$publisher_connstr' PUBLICATION alice WITH (password_required=false);
 ));
 
 # Wait for initial sync to finish