Add REPLICATION privilege for ROLEs

This privilege is required to do Streaming Replication, instead of
superuser, making it possible to set up a SR slave that doesn't
have write permissions on the master.

Superuser privileges do NOT override this check, so in order to
use the default superuser account for replication it must be
explicitly granted the REPLICATION permissions. This is backwards
incompatible change, in the interest of higher default security.

doc/src/sgml/catalogs.sgml

@@ -1235,6 +1235,17 @@
       </entry>
      </row>
 
+     <row>
+      <entry><structfield>rolreplication</structfield></entry>
+      <entry><type>bool</type></entry>
+      <entry>
+       Role is a replication role. That is, this role can initiate streaming
+       replication (see <xref linkend="streaming-replication">) and set/unset
+       the system backup mode using <function>pg_start_backup</> and
+       <function>pg_stop_backup</>.
+      </entry>
+     </row>
+
      <row>
       <entry><structfield>rolconnlimit</structfield></entry>
       <entry><type>int4</type></entry>

doc/src/sgml/func.sgml

@@ -13969,14 +13969,14 @@ SELECT set_config('log_statement_stats', 'off', false);
         <literal><function>pg_start_backup(<parameter>label</> <type>text</> <optional>, <parameter>fast</> <type>boolean</> </optional>)</function></literal>
         </entry>
        <entry><type>text</type></entry>
-       <entry>Prepare for performing on-line backup (restricted to superusers)</entry>
+       <entry>Prepare for performing on-line backup (restricted to superusers or replication roles)</entry>
       </row>
       <row>
        <entry>
         <literal><function>pg_stop_backup()</function></literal>
         </entry>
        <entry><type>text</type></entry>
-       <entry>Finish performing on-line backup (restricted to superusers)</entry>
+       <entry>Finish performing on-line backup (restricted to superusers or replication roles)</entry>
       </row>
       <row>
        <entry>

doc/src/sgml/high-availability.sgml

@@ -636,8 +636,8 @@ protocol to make nodes agree on a serializable transactional order.
    <para>
     If you want to use streaming replication, set up authentication on the
     primary server to allow replication connections from the standby
-    server(s); that is, provide a suitable entry or entries in
-    <filename>pg_hba.conf</> with the database field set to
+    server(s); that is, create a role and provide a suitable entry or
+    entries in <filename>pg_hba.conf</> with the database field set to
     <literal>replication</>.  Also ensure <varname>max_wal_senders</> is set
     to a sufficiently large value in the configuration file of the primary
     server.
@@ -796,15 +796,28 @@ archive_cleanup_command = 'pg_archivecleanup /path/to/archive %r'
      It is very important that the access privileges for replication be set up
      so that only trusted users can read the WAL stream, because it is
      easy to extract privileged information from it.  Standby servers must
-     authenticate to the primary as a superuser account.
-     So a role with the <literal>SUPERUSER</> and <literal>LOGIN</>
-     privileges needs to be created on the primary.
+     authenticate to the primary as an account that has the
+     <literal>REPLICATION</> privilege. So a role with the
+     <literal>REPLICATION</> and <literal>LOGIN</> privileges needs to be
+     created on the primary.
     </para>
+
+    <note>
+     <para>
+      It is recommended that a dedicated user account is used for replication.
+      While it is possible to add  the <literal>REPLICATION</> privilege to
+      a superuser account for the purporses of replication, this is not
+      recommended. While <literal>REPLICATION</> privilege gives very high
+      permissions, it does not allow the user to modify any data on the
+      primary system, which the <literal>SUPERUSER</> privilege does.
+     </para>
+    </note>
+
     <para>
      Client authentication for replication is controlled by a
      <filename>pg_hba.conf</> record specifying <literal>replication</> in the
      <replaceable>database</> field. For example, if the standby is running on
-     host IP <literal>192.168.1.100</> and the superuser's name for replication
+     host IP <literal>192.168.1.100</> and the account name for replication
      is <literal>foo</>, the administrator can add the following line to the
      <filename>pg_hba.conf</> file on the primary:
 
@@ -823,7 +836,7 @@ host    replication     foo             192.168.1.100/32        md5
      standby (specify <literal>replication</> in the <replaceable>database</>
      field).
      For example, if the primary is running on host IP <literal>192.168.1.50</>,
-     port <literal>5432</literal>, the superuser's name for replication is
+     port <literal>5432</literal>, the account name for replication is
      <literal>foo</>, and the password is <literal>foopass</>, the administrator
      can add the following line to the <filename>recovery.conf</> file on the
      standby:

doc/src/sgml/ref/alter_role.sgml

@@ -31,6 +31,7 @@ ALTER ROLE <replaceable class="PARAMETER">name</replaceable> [ [ WITH ] <replace
     | CREATEUSER | NOCREATEUSER
     | INHERIT | NOINHERIT
     | LOGIN | NOLOGIN
+    | REPLICATION | NOREPLICATION
     | CONNECTION LIMIT <replaceable class="PARAMETER">connlimit</replaceable>
     | [ ENCRYPTED | UNENCRYPTED ] PASSWORD '<replaceable class="PARAMETER">password</replaceable>'
     | VALID UNTIL '<replaceable class="PARAMETER">timestamp</replaceable>'
@@ -63,7 +64,7 @@ ALTER ROLE <replaceable class="PARAMETER">name</replaceable> [ IN DATABASE <repl
    Attributes not mentioned in the command retain their previous settings.
    Database superusers can change any of these settings for any role.
    Roles having <literal>CREATEROLE</> privilege can change any of these
-   settings, but only for non-superuser roles.
+   settings, but only for non-superuser and non-replication roles.
    Ordinary roles can only change their own password.
   </para>
 
@@ -127,6 +128,8 @@ ALTER ROLE <replaceable class="PARAMETER">name</replaceable> [ IN DATABASE <repl
       <term><literal>NOINHERIT</literal></term>
       <term><literal>LOGIN</literal></term>
       <term><literal>NOLOGIN</literal></term>
+      <term><literal>REPLICATION</literal></term>
+      <term><literal>NOREPLICATION</literal></term>
       <term><literal>CONNECTION LIMIT</literal> <replaceable class="parameter">connlimit</replaceable></term>
       <term><literal>PASSWORD</> <replaceable class="parameter">password</replaceable></term>
       <term><literal>ENCRYPTED</></term>

doc/src/sgml/ref/alter_user.sgml

@@ -31,6 +31,7 @@ ALTER USER <replaceable class="PARAMETER">name</replaceable> [ [ WITH ] <replace
     | CREATEUSER | NOCREATEUSER
     | INHERIT | NOINHERIT
     | LOGIN | NOLOGIN
+    | REPLICATION | NOREPLICATION
     | CONNECTION LIMIT <replaceable class="PARAMETER">connlimit</replaceable>
     | [ ENCRYPTED | UNENCRYPTED ] PASSWORD '<replaceable class="PARAMETER">password</replaceable>'
     | VALID UNTIL '<replaceable class="PARAMETER">timestamp</replaceable>'

doc/src/sgml/ref/create_role.sgml

@@ -31,6 +31,7 @@ CREATE ROLE <replaceable class="PARAMETER">name</replaceable> [ [ WITH ] <replac
     | CREATEUSER | NOCREATEUSER
     | INHERIT | NOINHERIT
     | LOGIN | NOLOGIN
+    | REPLICATION | NOREPLICATION
     | CONNECTION LIMIT <replaceable class="PARAMETER">connlimit</replaceable>
     | [ ENCRYPTED | UNENCRYPTED ] PASSWORD '<replaceable class="PARAMETER">password</replaceable>'
     | VALID UNTIL '<replaceable class="PARAMETER">timestamp</replaceable>'
@@ -174,6 +175,21 @@ CREATE ROLE <replaceable class="PARAMETER">name</replaceable> [ [ WITH ] <replac
       </listitem>
      </varlistentry>
 
+     <varlistentry>
+      <term><literal>REPLICATION</literal></term>
+      <term><literal>NOREPLICATION</literal></term>
+      <listitem>
+       <para>
+        These clauses determine whether a role is allowed to initiate
+        streaming replication or put the system in and out of backup mode.
+        A role having the <literal>REPLICATION</> attribute is a very
+        highly privileged role, and should only be used on roles actually
+	used for replication. If not specified,
+        <literal>NOREPLICATION</literal> is the default.
+       </para>
+      </listitem>
+     </varlistentry>
+
      <varlistentry>
       <term><literal>CONNECTION LIMIT</literal> <replaceable class="parameter">connlimit</replaceable></term>
       <listitem>

doc/src/sgml/ref/create_user.sgml

@@ -31,6 +31,7 @@ CREATE USER <replaceable class="PARAMETER">name</replaceable> [ [ WITH ] <replac
     | CREATEUSER | NOCREATEUSER
     | INHERIT | NOINHERIT
     | LOGIN | NOLOGIN
+    | REPLICATION | NOREPLICATION
     | CONNECTION LIMIT <replaceable class="PARAMETER">connlimit</replaceable>
     | [ ENCRYPTED | UNENCRYPTED ] PASSWORD '<replaceable class="PARAMETER">password</replaceable>'
     | VALID UNTIL '<replaceable class="PARAMETER">timestamp</replaceable>'

src/backend/access/transam/xlog.c

@@ -8301,10 +8301,10 @@ pg_start_backup(PG_FUNCTION_ARGS)
 	struct stat stat_buf;
 	FILE	   *fp;
 
-	if (!superuser())
+	if (!superuser() && !is_authenticated_user_replication_role())
 		ereport(ERROR,
 				(errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
-				 errmsg("must be superuser to run a backup")));
+				 errmsg("must be superuser or replication role to run a backup")));
 
 	if (RecoveryInProgress())
 		ereport(ERROR,
@@ -8493,10 +8493,10 @@ pg_stop_backup(PG_FUNCTION_ARGS)
 	int			waits = 0;
 	bool		reported_waiting = false;
 
-	if (!superuser())
+	if (!superuser() && !is_authenticated_user_replication_role())
 		ereport(ERROR,
 				(errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
-				 (errmsg("must be superuser to run a backup"))));
+				 (errmsg("must be superuser or replication role to run a backup"))));
 
 	if (RecoveryInProgress())
 		ereport(ERROR,

src/backend/catalog/system_views.sql

@@ -15,6 +15,7 @@ CREATE VIEW pg_roles AS
         rolcreatedb,
         rolcatupdate,
         rolcanlogin,
+        rolreplication,
         rolconnlimit,
         '********'::text as rolpassword,
         rolvaliduntil,
@@ -30,6 +31,7 @@ CREATE VIEW pg_shadow AS
         rolcreatedb AS usecreatedb,
         rolsuper AS usesuper,
         rolcatupdate AS usecatupd,
+        rolreplication AS userepl,
         rolpassword AS passwd,
         rolvaliduntil::abstime AS valuntil,
         setconfig AS useconfig
@@ -54,6 +56,7 @@ CREATE VIEW pg_user AS
         usecreatedb,
         usesuper,
         usecatupd,
+        userepl,
         '********'::text as passwd,
         valuntil,
         useconfig

src/backend/commands/user.c

@@ -94,6 +94,7 @@ CreateRole(CreateRoleStmt *stmt)
 	bool		createrole = false;		/* Can this user create roles? */
 	bool		createdb = false;		/* Can the user create databases? */
 	bool		canlogin = false;		/* Can this user login? */
+	bool		isreplication = false; /* Is this a replication role? */
 	int			connlimit = -1; /* maximum connections allowed */
 	List	   *addroleto = NIL;	/* roles to make this a member of */
 	List	   *rolemembers = NIL;		/* roles to be members of this role */
@@ -107,6 +108,7 @@ CreateRole(CreateRoleStmt *stmt)
 	DefElem    *dcreaterole = NULL;
 	DefElem    *dcreatedb = NULL;
 	DefElem    *dcanlogin = NULL;
+	DefElem	   *disreplication = NULL;
 	DefElem    *dconnlimit = NULL;
 	DefElem    *daddroleto = NULL;
 	DefElem    *drolemembers = NULL;
@@ -190,6 +192,14 @@ CreateRole(CreateRoleStmt *stmt)
 						 errmsg("conflicting or redundant options")));
 			dcanlogin = defel;
 		}
+		else if (strcmp(defel->defname, "isreplication") == 0)
+		{
+			if (disreplication)
+				ereport(ERROR,
+						(errcode(ERRCODE_SYNTAX_ERROR),
+						 errmsg("conflicting or redundant options")));
+			disreplication = defel;
+		}
 		else if (strcmp(defel->defname, "connectionlimit") == 0)
 		{
 			if (dconnlimit)
@@ -247,6 +257,8 @@ CreateRole(CreateRoleStmt *stmt)
 		createdb = intVal(dcreatedb->arg) != 0;
 	if (dcanlogin)
 		canlogin = intVal(dcanlogin->arg) != 0;
+	if (disreplication)
+		isreplication = intVal(disreplication->arg) != 0;
 	if (dconnlimit)
 	{
 		connlimit = intVal(dconnlimit->arg);
@@ -272,6 +284,13 @@ CreateRole(CreateRoleStmt *stmt)
 					(errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
 					 errmsg("must be superuser to create superusers")));
 	}
+	else if (isreplication)
+	{
+		if (!superuser())
+			ereport(ERROR,
+					(errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
+					 errmsg("must be superuser to create replication users")));
+	}
 	else
 	{
 		if (!have_createrole_privilege())
@@ -341,6 +360,7 @@ CreateRole(CreateRoleStmt *stmt)
 	/* superuser gets catupdate right by default */
 	new_record[Anum_pg_authid_rolcatupdate - 1] = BoolGetDatum(issuper);
 	new_record[Anum_pg_authid_rolcanlogin - 1] = BoolGetDatum(canlogin);
+	new_record[Anum_pg_authid_rolreplication - 1] = BoolGetDatum(isreplication);
 	new_record[Anum_pg_authid_rolconnlimit - 1] = Int32GetDatum(connlimit);
 
 	if (password)
@@ -439,6 +459,7 @@ AlterRole(AlterRoleStmt *stmt)
 	int			createrole = -1;	/* Can this user create roles? */
 	int			createdb = -1;	/* Can the user create databases? */
 	int			canlogin = -1;	/* Can this user login? */
+	int			isreplication = -1; /* Is this a replication role? */
 	int			connlimit = -1; /* maximum connections allowed */
 	List	   *rolemembers = NIL;		/* roles to be added/removed */
 	char	   *validUntil = NULL;		/* time the login is valid until */
@@ -450,6 +471,7 @@ AlterRole(AlterRoleStmt *stmt)
 	DefElem    *dcreaterole = NULL;
 	DefElem    *dcreatedb = NULL;
 	DefElem    *dcanlogin = NULL;
+	DefElem	   *disreplication = NULL;
 	DefElem    *dconnlimit = NULL;
 	DefElem    *drolemembers = NULL;
 	DefElem    *dvalidUntil = NULL;
@@ -514,6 +536,14 @@ AlterRole(AlterRoleStmt *stmt)
 						 errmsg("conflicting or redundant options")));
 			dcanlogin = defel;
 		}
+		else if (strcmp(defel->defname, "isreplication") == 0)
+		{
+			if (disreplication)
+				ereport(ERROR,
+						(errcode(ERRCODE_SYNTAX_ERROR),
+						 errmsg("conflicting or redundant options")));
+			disreplication = defel;
+		}
 		else if (strcmp(defel->defname, "connectionlimit") == 0)
 		{
 			if (dconnlimit)
@@ -556,6 +586,8 @@ AlterRole(AlterRoleStmt *stmt)
 		createdb = intVal(dcreatedb->arg);
 	if (dcanlogin)
 		canlogin = intVal(dcanlogin->arg);
+	if (disreplication)
+		isreplication = intVal(disreplication->arg);
 	if (dconnlimit)
 	{
 		connlimit = intVal(dconnlimit->arg);
@@ -594,12 +626,20 @@ AlterRole(AlterRoleStmt *stmt)
 					(errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
 					 errmsg("must be superuser to alter superusers")));
 	}
+	else if (((Form_pg_authid) GETSTRUCT(tuple))->rolreplication || isreplication >= 0)
+	{
+		if (!superuser())
+			ereport(ERROR,
+					(errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
+					 errmsg("must be superuser to alter replication users")));
+	}
 	else if (!have_createrole_privilege())
 	{
 		if (!(inherit < 0 &&
 			  createrole < 0 &&
 			  createdb < 0 &&
 			  canlogin < 0 &&
+			  isreplication < 0 &&
 			  !dconnlimit &&
 			  !rolemembers &&
 			  !validUntil &&
@@ -685,6 +725,12 @@ AlterRole(AlterRoleStmt *stmt)
 		new_record_repl[Anum_pg_authid_rolcanlogin - 1] = true;
 	}
 
+	if (isreplication >= 0)
+	{
+		new_record[Anum_pg_authid_rolreplication - 1] = BoolGetDatum(isreplication > 0);
+		new_record_repl[Anum_pg_authid_rolreplication - 1] = true;
+	}
+
 	if (dconnlimit)
 	{
 		new_record[Anum_pg_authid_rolconnlimit - 1] = Int32GetDatum(connlimit);

src/backend/parser/gram.y

@@ -510,8 +510,9 @@ static RangeVar *makeRangeVarFromAnyName(List *names, int position, core_yyscan_
 	MAPPING MATCH MAXVALUE MINUTE_P MINVALUE MODE MONTH_P MOVE
 
 	NAME_P NAMES NATIONAL NATURAL NCHAR NEXT NO NOCREATEDB
-	NOCREATEROLE NOCREATEUSER NOINHERIT NOLOGIN_P NONE NOSUPERUSER
-	NOT NOTHING NOTIFY NOTNULL NOWAIT NULL_P NULLIF NULLS_P NUMERIC
+	NOCREATEROLE NOCREATEUSER NOINHERIT NOLOGIN_P NONE NOREPLICATION_P
+	NOSUPERUSER NOT NOTHING NOTIFY NOTNULL NOWAIT NULL_P NULLIF
+	NULLS_P NUMERIC
 
 	OBJECT_P OF OFF OFFSET OIDS ON ONLY OPERATOR OPTION OPTIONS OR
 	ORDER OUT_P OUTER_P OVER OVERLAPS OVERLAY OWNED OWNER
@@ -523,8 +524,9 @@ static RangeVar *makeRangeVarFromAnyName(List *names, int position, core_yyscan_
 	QUOTE
 
 	RANGE READ REAL REASSIGN RECHECK RECURSIVE REF REFERENCES REINDEX
-	RELATIVE_P RELEASE RENAME REPEATABLE REPLACE REPLICA RESET RESTART
-	RESTRICT RETURNING RETURNS REVOKE RIGHT ROLE ROLLBACK ROW ROWS RULE
+	RELATIVE_P RELEASE RENAME REPEATABLE REPLACE REPLICA REPLICATION_P
+	RESET RESTART RESTRICT RETURNING RETURNS REVOKE RIGHT ROLE ROLLBACK
+	ROW ROWS RULE
 
 	SAVEPOINT SCHEMA SCROLL SEARCH SECOND_P SECURITY SELECT SEQUENCE SEQUENCES
 	SERIALIZABLE SERVER SESSION SESSION_USER SET SETOF SHARE
@@ -864,6 +866,14 @@ AlterOptRoleElem:
 				{
 					$$ = makeDefElem("canlogin", (Node *)makeInteger(FALSE));
 				}
+			| REPLICATION_P
+				{
+					$$ = makeDefElem("isreplication", (Node *)makeInteger(TRUE));
+				}
+			| NOREPLICATION_P
+				{
+					$$ = makeDefElem("isreplication", (Node *)makeInteger(FALSE));
+				}
 			| CONNECTION LIMIT SignedIconst
 				{
 					$$ = makeDefElem("connectionlimit", (Node *)makeInteger($3));
@@ -11288,6 +11298,7 @@ unreserved_keyword:
 			| NOCREATEUSER
 			| NOINHERIT
 			| NOLOGIN_P
+			| NOREPLICATION_P
 			| NOSUPERUSER
 			| NOTHING
 			| NOTIFY
@@ -11330,6 +11341,7 @@ unreserved_keyword:
 			| REPEATABLE
 			| REPLACE
 			| REPLICA
+			| REPLICATION_P
 			| RESET
 			| RESTART
 			| RESTRICT

src/backend/utils/init/miscinit.c

@@ -231,6 +231,7 @@ static int	SecurityRestrictionContext = 0;
 static bool SetRoleIsActive = false;
 
 
+
 /*
  * GetUserId - get the current effective user ID.
  *
@@ -388,6 +389,24 @@ SetUserIdAndContext(Oid userid, bool sec_def_context)
 }
 
 
+/*
+ * Check if the authenticated user is a replication role
+ */
+bool
+is_authenticated_user_replication_role(void)
+{
+	bool            result = false;
+	HeapTuple       utup;
+
+	utup = SearchSysCache1(AUTHOID, ObjectIdGetDatum(AuthenticatedUserId));
+	if (HeapTupleIsValid(utup))
+	{
+		result = ((Form_pg_authid) GETSTRUCT(utup))->rolreplication;
+		ReleaseSysCache(utup);
+	}
+	return result;
+}
+
 /*
  * Initialize user identity during normal backend startup
  */

src/backend/utils/init/postinit.c

@@ -658,11 +658,11 @@ InitPostgres(const char *in_dbname, Oid dboid, const char *username,
 	{
 		Assert(!bootstrap);
 
-		/* must have authenticated as a superuser */
-		if (!am_superuser)
+		/* must have authenticated as a replication role */
+		if (!is_authenticated_user_replication_role())
 			ereport(FATAL,
 					(errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
-					 errmsg("must be superuser to start walsender")));
+					 errmsg("must be replication role to start walsender")));
 
 		/* process any options passed in the startup packet */
 		if (MyProcPort != NULL)

src/bin/pg_dump/pg_dumpall.c

@@ -653,16 +653,26 @@ dumpRoles(PGconn *conn)
 				i_rolconnlimit,
 				i_rolpassword,
 				i_rolvaliduntil,
+				i_rolreplication,
 				i_rolcomment;
 	int			i;
 
 	/* note: rolconfig is dumped later */
-	if (server_version >= 80200)
+	if (server_version >= 90100)
 		printfPQExpBuffer(buf,
 						  "SELECT rolname, rolsuper, rolinherit, "
 						  "rolcreaterole, rolcreatedb, rolcatupdate, "
 						  "rolcanlogin, rolconnlimit, rolpassword, "
-						  "rolvaliduntil, "
+						  "rolvaliduntil, rolreplication, "
+			  "pg_catalog.shobj_description(oid, 'pg_authid') as rolcomment "
+						  "FROM pg_authid "
+						  "ORDER BY 1");
+	else if (server_version >= 80200)
+		printfPQExpBuffer(buf,
+						  "SELECT rolname, rolsuper, rolinherit, "
+						  "rolcreaterole, rolcreatedb, rolcatupdate, "
+						  "rolcanlogin, rolconnlimit, rolpassword, "
+						  "rolvaliduntil, false as rolreplication, "
 			  "pg_catalog.shobj_description(oid, 'pg_authid') as rolcomment "
 						  "FROM pg_authid "
 						  "ORDER BY 1");
@@ -671,7 +681,8 @@ dumpRoles(PGconn *conn)
 						  "SELECT rolname, rolsuper, rolinherit, "
 						  "rolcreaterole, rolcreatedb, rolcatupdate, "
 						  "rolcanlogin, rolconnlimit, rolpassword, "
-						  "rolvaliduntil, null as rolcomment "
+						  "rolvaliduntil, false as rolreplication, "
+						  "null as rolcomment "
 						  "FROM pg_authid "
 						  "ORDER BY 1");
 	else
@@ -686,6 +697,7 @@ dumpRoles(PGconn *conn)
 						  "-1 as rolconnlimit, "
 						  "passwd as rolpassword, "
 						  "valuntil as rolvaliduntil, "
+						  "false as rolreplication, "
 						  "null as rolcomment "
 						  "FROM pg_shadow "
 						  "UNION ALL "
@@ -699,6 +711,7 @@ dumpRoles(PGconn *conn)
 						  "-1 as rolconnlimit, "
 						  "null::text as rolpassword, "
 						  "null::abstime as rolvaliduntil, "
+						  "false as rolreplication, "
 						  "null as rolcomment "
 						  "FROM pg_group "
 						  "WHERE NOT EXISTS (SELECT 1 FROM pg_shadow "
@@ -717,6 +730,7 @@ dumpRoles(PGconn *conn)
 	i_rolconnlimit = PQfnumber(res, "rolconnlimit");
 	i_rolpassword = PQfnumber(res, "rolpassword");
 	i_rolvaliduntil = PQfnumber(res, "rolvaliduntil");
+	i_rolreplication = PQfnumber(res, "rolreplication");
 	i_rolcomment = PQfnumber(res, "rolcomment");
 
 	if (PQntuples(res) > 0)
@@ -765,6 +779,11 @@ dumpRoles(PGconn *conn)
 		else
 			appendPQExpBuffer(buf, " NOLOGIN");
 
+		if (strcmp(PQgetvalue(res, i, i_rolreplication), "t") == 0)
+			appendPQExpBuffer(buf, " REPLICATION");
+		else
+			appendPQExpBuffer(buf, " NOREPLICATION");
+
 		if (strcmp(PQgetvalue(res, i, i_rolconnlimit), "-1") != 0)
 			appendPQExpBuffer(buf, " CONNECTION LIMIT %s",
 							  PQgetvalue(res, i, i_rolconnlimit));

src/bin/psql/describe.c

@@ -2195,6 +2195,10 @@ describeRoles(const char *pattern, bool verbose)
 			appendPQExpBufferStr(&buf, "\n, pg_catalog.shobj_description(r.oid, 'pg_authid') AS description");
 			ncols++;
 		}
+		if (pset.sversion >= 90100)
+		{
+			appendPQExpBufferStr(&buf,"\n, r.rolreplication");
+		}
 
 		appendPQExpBufferStr(&buf, "\nFROM pg_catalog.pg_roles r\n");
 
@@ -2254,6 +2258,10 @@ describeRoles(const char *pattern, bool verbose)
 		if (strcmp(PQgetvalue(res, i, 5), "t") != 0)
 			add_role_attribute(&buf, _("Cannot login"));
 
+		if (pset.sversion >= 90100)
+			if (strcmp(PQgetvalue(res, i, 8), "t") == 0)
+				add_role_attribute(&buf, _("Replication"));
+
 		conns = atoi(PQgetvalue(res, i, 6));
 		if (conns >= 0)
 		{

src/include/catalog/pg_authid.h

@@ -51,6 +51,7 @@ CATALOG(pg_authid,1260) BKI_SHARED_RELATION BKI_ROWTYPE_OID(2842) BKI_SCHEMA_MAC
 	bool		rolcreatedb;	/* allowed to create databases? */
 	bool		rolcatupdate;	/* allowed to alter catalogs manually? */
 	bool		rolcanlogin;	/* allowed to log in as session user? */
+	bool        rolreplication; /* role used for streaming replication */
 	int4		rolconnlimit;	/* max connections allowed (-1=no limit) */
 
 	/* remaining fields may be null; use heap_getattr to read them! */
@@ -72,7 +73,7 @@ typedef FormData_pg_authid *Form_pg_authid;
  *		compiler constants for pg_authid
  * ----------------
  */
-#define Natts_pg_authid					10
+#define Natts_pg_authid					11
 #define Anum_pg_authid_rolname			1
 #define Anum_pg_authid_rolsuper			2
 #define Anum_pg_authid_rolinherit		3
@@ -80,9 +81,10 @@ typedef FormData_pg_authid *Form_pg_authid;
 #define Anum_pg_authid_rolcreatedb		5
 #define Anum_pg_authid_rolcatupdate		6
 #define Anum_pg_authid_rolcanlogin		7
-#define Anum_pg_authid_rolconnlimit		8
-#define Anum_pg_authid_rolpassword		9
-#define Anum_pg_authid_rolvaliduntil	10
+#define Anum_pg_authid_rolreplication	8
+#define Anum_pg_authid_rolconnlimit		9
+#define Anum_pg_authid_rolpassword		10
+#define Anum_pg_authid_rolvaliduntil	11
 
 /* ----------------
  *		initial contents of pg_authid
@@ -91,7 +93,7 @@ typedef FormData_pg_authid *Form_pg_authid;
  * user choices.
  * ----------------
  */
-DATA(insert OID = 10 ( "POSTGRES" t t t t t t -1 _null_ _null_ ));
+DATA(insert OID = 10 ( "POSTGRES" t t t t t t f -1 _null_ _null_ ));
 
 #define BOOTSTRAP_SUPERUSERID 10
 

src/include/miscadmin.h

@@ -357,6 +357,7 @@ extern void ValidatePgVersion(const char *path);
 extern void process_shared_preload_libraries(void);
 extern void process_local_preload_libraries(void);
 extern void pg_bindtextdomain(const char *domain);
+extern bool is_authenticated_user_replication_role(void);
 
 /* in access/transam/xlog.c */
 extern bool BackupInProgress(void);

src/include/parser/kwlist.h

@@ -250,6 +250,7 @@ PG_KEYWORD("nocreateuser", NOCREATEUSER, UNRESERVED_KEYWORD)
 PG_KEYWORD("noinherit", NOINHERIT, UNRESERVED_KEYWORD)
 PG_KEYWORD("nologin", NOLOGIN_P, UNRESERVED_KEYWORD)
 PG_KEYWORD("none", NONE, COL_NAME_KEYWORD)
+PG_KEYWORD("noreplication", NOREPLICATION_P, UNRESERVED_KEYWORD)
 PG_KEYWORD("nosuperuser", NOSUPERUSER, UNRESERVED_KEYWORD)
 PG_KEYWORD("not", NOT, RESERVED_KEYWORD)
 PG_KEYWORD("nothing", NOTHING, UNRESERVED_KEYWORD)
@@ -313,6 +314,7 @@ PG_KEYWORD("rename", RENAME, UNRESERVED_KEYWORD)
 PG_KEYWORD("repeatable", REPEATABLE, UNRESERVED_KEYWORD)
 PG_KEYWORD("replace", REPLACE, UNRESERVED_KEYWORD)
 PG_KEYWORD("replica", REPLICA, UNRESERVED_KEYWORD)
+PG_KEYWORD("replication", REPLICATION_P, UNRESERVED_KEYWORD)
 PG_KEYWORD("reset", RESET, UNRESERVED_KEYWORD)
 PG_KEYWORD("restart", RESTART, UNRESERVED_KEYWORD)
 PG_KEYWORD("restrict", RESTRICT, UNRESERVED_KEYWORD)

src/test/regress/expected/rules.out

@@ -1285,11 +1285,11 @@ SELECT viewname, definition FROM pg_views WHERE schemaname <> 'information_schem
  pg_locks                    | SELECT l.locktype, l.database, l.relation, l.page, l.tuple, l.virtualxid, l.transactionid, l.classid, l.objid, l.objsubid, l.virtualtransaction, l.pid, l.mode, l.granted FROM pg_lock_status() l(locktype, database, relation, page, tuple, virtualxid, transactionid, classid, objid, objsubid, virtualtransaction, pid, mode, granted);
  pg_prepared_statements      | SELECT p.name, p.statement, p.prepare_time, p.parameter_types, p.from_sql FROM pg_prepared_statement() p(name, statement, prepare_time, parameter_types, from_sql);
  pg_prepared_xacts           | SELECT p.transaction, p.gid, p.prepared, u.rolname AS owner, d.datname AS database FROM ((pg_prepared_xact() p(transaction, gid, prepared, ownerid, dbid) LEFT JOIN pg_authid u ON ((p.ownerid = u.oid))) LEFT JOIN pg_database d ON ((p.dbid = d.oid)));
- pg_roles                    | SELECT pg_authid.rolname, pg_authid.rolsuper, pg_authid.rolinherit, pg_authid.rolcreaterole, pg_authid.rolcreatedb, pg_authid.rolcatupdate, pg_authid.rolcanlogin, pg_authid.rolconnlimit, '********'::text AS rolpassword, pg_authid.rolvaliduntil, s.setconfig AS rolconfig, pg_authid.oid FROM (pg_authid LEFT JOIN pg_db_role_setting s ON (((pg_authid.oid = s.setrole) AND (s.setdatabase = (0)::oid))));
+ pg_roles                    | SELECT pg_authid.rolname, pg_authid.rolsuper, pg_authid.rolinherit, pg_authid.rolcreaterole, pg_authid.rolcreatedb, pg_authid.rolcatupdate, pg_authid.rolcanlogin, pg_authid.rolreplication, pg_authid.rolconnlimit, '********'::text AS rolpassword, pg_authid.rolvaliduntil, s.setconfig AS rolconfig, pg_authid.oid FROM (pg_authid LEFT JOIN pg_db_role_setting s ON (((pg_authid.oid = s.setrole) AND (s.setdatabase = (0)::oid))));
  pg_rules                    | SELECT n.nspname AS schemaname, c.relname AS tablename, r.rulename, pg_get_ruledef(r.oid) AS definition FROM ((pg_rewrite r JOIN pg_class c ON ((c.oid = r.ev_class))) LEFT JOIN pg_namespace n ON ((n.oid = c.relnamespace))) WHERE (r.rulename <> '_RETURN'::name);
  pg_seclabels                | (((((SELECT l.objoid, l.classoid, l.objsubid, CASE WHEN (rel.relkind = 'r'::"char") THEN 'table'::text WHEN (rel.relkind = 'v'::"char") THEN 'view'::text WHEN (rel.relkind = 'S'::"char") THEN 'sequence'::text ELSE NULL::text END AS objtype, rel.relnamespace AS objnamespace, CASE WHEN pg_table_is_visible(rel.oid) THEN quote_ident((rel.relname)::text) ELSE ((quote_ident((nsp.nspname)::text) || '.'::text) || quote_ident((rel.relname)::text)) END AS objname, l.provider, l.label FROM ((pg_seclabel l JOIN pg_class rel ON (((l.classoid = rel.tableoid) AND (l.objoid = rel.oid)))) JOIN pg_namespace nsp ON ((rel.relnamespace = nsp.oid))) WHERE (l.objsubid = 0) UNION ALL SELECT l.objoid, l.classoid, l.objsubid, 'column'::text AS objtype, rel.relnamespace AS objnamespace, ((CASE WHEN pg_table_is_visible(rel.oid) THEN quote_ident((rel.relname)::text) ELSE ((quote_ident((nsp.nspname)::text) || '.'::text) || quote_ident((rel.relname)::text)) END || '.'::text) || (att.attname)::text) AS objname, l.provider, l.label FROM (((pg_seclabel l JOIN pg_class rel ON (((l.classoid = rel.tableoid) AND (l.objoid = rel.oid)))) JOIN pg_attribute att ON (((rel.oid = att.attrelid) AND (l.objsubid = att.attnum)))) JOIN pg_namespace nsp ON ((rel.relnamespace = nsp.oid))) WHERE (l.objsubid <> 0)) UNION ALL SELECT l.objoid, l.classoid, l.objsubid, CASE WHEN (pro.proisagg = true) THEN 'aggregate'::text WHEN (pro.proisagg = false) THEN 'function'::text ELSE NULL::text END AS objtype, pro.pronamespace AS objnamespace, (((CASE WHEN pg_function_is_visible(pro.oid) THEN quote_ident((pro.proname)::text) ELSE ((quote_ident((nsp.nspname)::text) || '.'::text) || quote_ident((pro.proname)::text)) END || '('::text) || pg_get_function_arguments(pro.oid)) || ')'::text) AS objname, l.provider, l.label FROM ((pg_seclabel l JOIN pg_proc pro ON (((l.classoid = pro.tableoid) AND (l.objoid = pro.oid)))) JOIN pg_namespace nsp ON ((pro.pronamespace = nsp.oid))) WHERE (l.objsubid = 0)) UNION ALL SELECT l.objoid, l.classoid, l.objsubid, CASE WHEN (typ.typtype = 'd'::"char") THEN 'domain'::text ELSE 'type'::text END AS objtype, typ.typnamespace AS objnamespace, CASE WHEN pg_type_is_visible(typ.oid) THEN quote_ident((typ.typname)::text) ELSE ((quote_ident((nsp.nspname)::text) || '.'::text) || quote_ident((typ.typname)::text)) END AS objname, l.provider, l.label FROM ((pg_seclabel l JOIN pg_type typ ON (((l.classoid = typ.tableoid) AND (l.objoid = typ.oid)))) JOIN pg_namespace nsp ON ((typ.typnamespace = nsp.oid))) WHERE (l.objsubid = 0)) UNION ALL SELECT l.objoid, l.classoid, l.objsubid, 'large object'::text AS objtype, NULL::oid AS objnamespace, (l.objoid)::text AS objname, l.provider, l.label FROM (pg_seclabel l JOIN pg_largeobject_metadata lom ON ((l.objoid = lom.oid))) WHERE ((l.classoid = ('pg_largeobject'::regclass)::oid) AND (l.objsubid = 0))) UNION ALL SELECT l.objoid, l.classoid, l.objsubid, 'language'::text AS objtype, NULL::oid AS objnamespace, quote_ident((lan.lanname)::text) AS objname, l.provider, l.label FROM (pg_seclabel l JOIN pg_language lan ON (((l.classoid = lan.tableoid) AND (l.objoid = lan.oid)))) WHERE (l.objsubid = 0)) UNION ALL SELECT l.objoid, l.classoid, l.objsubid, 'schema'::text AS objtype, nsp.oid AS objnamespace, quote_ident((nsp.nspname)::text) AS objname, l.provider, l.label FROM (pg_seclabel l JOIN pg_namespace nsp ON (((l.classoid = nsp.tableoid) AND (l.objoid = nsp.oid)))) WHERE (l.objsubid = 0);
  pg_settings                 | SELECT a.name, a.setting, a.unit, a.category, a.short_desc, a.extra_desc, a.context, a.vartype, a.source, a.min_val, a.max_val, a.enumvals, a.boot_val, a.reset_val, a.sourcefile, a.sourceline FROM pg_show_all_settings() a(name, setting, unit, category, short_desc, extra_desc, context, vartype, source, min_val, max_val, enumvals, boot_val, reset_val, sourcefile, sourceline);
- pg_shadow                   | SELECT pg_authid.rolname AS usename, pg_authid.oid AS usesysid, pg_authid.rolcreatedb AS usecreatedb, pg_authid.rolsuper AS usesuper, pg_authid.rolcatupdate AS usecatupd, pg_authid.rolpassword AS passwd, (pg_authid.rolvaliduntil)::abstime AS valuntil, s.setconfig AS useconfig FROM (pg_authid LEFT JOIN pg_db_role_setting s ON (((pg_authid.oid = s.setrole) AND (s.setdatabase = (0)::oid)))) WHERE pg_authid.rolcanlogin;
+ pg_shadow                   | SELECT pg_authid.rolname AS usename, pg_authid.oid AS usesysid, pg_authid.rolcreatedb AS usecreatedb, pg_authid.rolsuper AS usesuper, pg_authid.rolcatupdate AS usecatupd, pg_authid.rolreplication AS userepl, pg_authid.rolpassword AS passwd, (pg_authid.rolvaliduntil)::abstime AS valuntil, s.setconfig AS useconfig FROM (pg_authid LEFT JOIN pg_db_role_setting s ON (((pg_authid.oid = s.setrole) AND (s.setdatabase = (0)::oid)))) WHERE pg_authid.rolcanlogin;
  pg_stat_activity            | SELECT s.datid, d.datname, s.procpid, s.usesysid, u.rolname AS usename, s.application_name, s.client_addr, s.client_port, s.backend_start, s.xact_start, s.query_start, s.waiting, s.current_query FROM pg_database d, pg_stat_get_activity(NULL::integer) s(datid, procpid, usesysid, application_name, current_query, waiting, xact_start, query_start, backend_start, client_addr, client_port), pg_authid u WHERE ((s.datid = d.oid) AND (s.usesysid = u.oid));
  pg_stat_all_indexes         | SELECT c.oid AS relid, i.oid AS indexrelid, n.nspname AS schemaname, c.relname, i.relname AS indexrelname, pg_stat_get_numscans(i.oid) AS idx_scan, pg_stat_get_tuples_returned(i.oid) AS idx_tup_read, pg_stat_get_tuples_fetched(i.oid) AS idx_tup_fetch FROM (((pg_class c JOIN pg_index x ON ((c.oid = x.indrelid))) JOIN pg_class i ON ((i.oid = x.indexrelid))) LEFT JOIN pg_namespace n ON ((n.oid = c.relnamespace))) WHERE (c.relkind = ANY (ARRAY['r'::"char", 't'::"char"]));
  pg_stat_all_tables          | SELECT c.oid AS relid, n.nspname AS schemaname, c.relname, pg_stat_get_numscans(c.oid) AS seq_scan, pg_stat_get_tuples_returned(c.oid) AS seq_tup_read, (sum(pg_stat_get_numscans(i.indexrelid)))::bigint AS idx_scan, ((sum(pg_stat_get_tuples_fetched(i.indexrelid)))::bigint + pg_stat_get_tuples_fetched(c.oid)) AS idx_tup_fetch, pg_stat_get_tuples_inserted(c.oid) AS n_tup_ins, pg_stat_get_tuples_updated(c.oid) AS n_tup_upd, pg_stat_get_tuples_deleted(c.oid) AS n_tup_del, pg_stat_get_tuples_hot_updated(c.oid) AS n_tup_hot_upd, pg_stat_get_live_tuples(c.oid) AS n_live_tup, pg_stat_get_dead_tuples(c.oid) AS n_dead_tup, pg_stat_get_last_vacuum_time(c.oid) AS last_vacuum, pg_stat_get_last_autovacuum_time(c.oid) AS last_autovacuum, pg_stat_get_last_analyze_time(c.oid) AS last_analyze, pg_stat_get_last_autoanalyze_time(c.oid) AS last_autoanalyze, pg_stat_get_vacuum_count(c.oid) AS vacuum_count, pg_stat_get_autovacuum_count(c.oid) AS autovacuum_count, pg_stat_get_analyze_count(c.oid) AS analyze_count, pg_stat_get_autoanalyze_count(c.oid) AS autoanalyze_count FROM ((pg_class c LEFT JOIN pg_index i ON ((c.oid = i.indrelid))) LEFT JOIN pg_namespace n ON ((n.oid = c.relnamespace))) WHERE (c.relkind = ANY (ARRAY['r'::"char", 't'::"char"])) GROUP BY c.oid, n.nspname, c.relname;
@@ -1317,7 +1317,7 @@ SELECT viewname, definition FROM pg_views WHERE schemaname <> 'information_schem
  pg_tables                   | SELECT n.nspname AS schemaname, c.relname AS tablename, pg_get_userbyid(c.relowner) AS tableowner, t.spcname AS tablespace, c.relhasindex AS hasindexes, c.relhasrules AS hasrules, c.relhastriggers AS hastriggers FROM ((pg_class c LEFT JOIN pg_namespace n ON ((n.oid = c.relnamespace))) LEFT JOIN pg_tablespace t ON ((t.oid = c.reltablespace))) WHERE (c.relkind = 'r'::"char");
  pg_timezone_abbrevs         | SELECT pg_timezone_abbrevs.abbrev, pg_timezone_abbrevs.utc_offset, pg_timezone_abbrevs.is_dst FROM pg_timezone_abbrevs() pg_timezone_abbrevs(abbrev, utc_offset, is_dst);
  pg_timezone_names           | SELECT pg_timezone_names.name, pg_timezone_names.abbrev, pg_timezone_names.utc_offset, pg_timezone_names.is_dst FROM pg_timezone_names() pg_timezone_names(name, abbrev, utc_offset, is_dst);
- pg_user                     | SELECT pg_shadow.usename, pg_shadow.usesysid, pg_shadow.usecreatedb, pg_shadow.usesuper, pg_shadow.usecatupd, '********'::text AS passwd, pg_shadow.valuntil, pg_shadow.useconfig FROM pg_shadow;
+ pg_user                     | SELECT pg_shadow.usename, pg_shadow.usesysid, pg_shadow.usecreatedb, pg_shadow.usesuper, pg_shadow.usecatupd, pg_shadow.userepl, '********'::text AS passwd, pg_shadow.valuntil, pg_shadow.useconfig FROM pg_shadow;
  pg_user_mappings            | SELECT u.oid AS umid, s.oid AS srvid, s.srvname, u.umuser, CASE WHEN (u.umuser = (0)::oid) THEN 'public'::name ELSE a.rolname END AS usename, CASE WHEN (pg_has_role(s.srvowner, 'USAGE'::text) OR has_server_privilege(s.oid, 'USAGE'::text)) THEN u.umoptions ELSE NULL::text[] END AS umoptions FROM ((pg_user_mapping u LEFT JOIN pg_authid a ON ((a.oid = u.umuser))) JOIN pg_foreign_server s ON ((u.umserver = s.oid)));
  pg_views                    | SELECT n.nspname AS schemaname, c.relname AS viewname, pg_get_userbyid(c.relowner) AS viewowner, pg_get_viewdef(c.oid) AS definition FROM (pg_class c LEFT JOIN pg_namespace n ON ((n.oid = c.relnamespace))) WHERE (c.relkind = 'v'::"char");
  rtest_v1                    | SELECT rtest_t1.a, rtest_t1.b FROM rtest_t1;