Support SECURITY LABEL on databases, tablespaces, and roles.

This requires a new shared catalog, pg_shseclabel.

Along the way, fix the security_label regression tests so that they
don't monkey with the labels of any pre-existing objects.  This is
unlikely to matter in practice, since only the label for the "dummy"
provider was being manipulated.  But this way still seems cleaner.

KaiGai Kohei, with fairly extensive hacking by me.

8 files changed, 225 insertions, 11 deletions

diff --git a/src/backend/catalog/Makefile b/src/backend/catalog/Makefile
index 3a834618d28..7e0b7d65a87 100644
--- a/src/backend/catalog/Makefile
+++ b/src/backend/catalog/Makefile
@@ -39,7 +39,7 @@ POSTGRES_BKI_SRCS = $(addprefix $(top_srcdir)/src/include/catalog/,\
 	pg_ts_parser.h pg_ts_template.h pg_extension.h \
 	pg_foreign_data_wrapper.h pg_foreign_server.h pg_user_mapping.h \
 	pg_foreign_table.h \
-	pg_default_acl.h pg_seclabel.h pg_collation.h \
+	pg_default_acl.h pg_seclabel.h pg_shseclabel.h pg_collation.h \
 	toasting.h indexing.h \
     )
 
diff --git a/src/backend/catalog/catalog.c b/src/backend/catalog/catalog.c
index cbce0072de1..9d568b5273a 100644
--- a/src/backend/catalog/catalog.c
+++ b/src/backend/catalog/catalog.c
@@ -34,6 +34,7 @@
 #include "catalog/pg_db_role_setting.h"
 #include "catalog/pg_shdepend.h"
 #include "catalog/pg_shdescription.h"
+#include "catalog/pg_shseclabel.h"
 #include "catalog/pg_tablespace.h"
 #include "catalog/toasting.h"
 #include "miscadmin.h"
@@ -380,6 +381,7 @@ IsSharedRelation(Oid relationId)
 		relationId == PLTemplateRelationId ||
 		relationId == SharedDescriptionRelationId ||
 		relationId == SharedDependRelationId ||
+		relationId == SharedSecLabelRelationId ||
 		relationId == TableSpaceRelationId ||
 		relationId == DbRoleSettingRelationId)
 		return true;
@@ -394,6 +396,7 @@ IsSharedRelation(Oid relationId)
 		relationId == SharedDescriptionObjIndexId ||
 		relationId == SharedDependDependerIndexId ||
 		relationId == SharedDependReferenceIndexId ||
+		relationId == SharedSecLabelObjectIndexId ||
 		relationId == TablespaceOidIndexId ||
 		relationId == TablespaceNameIndexId ||
 		relationId == DbRoleSettingDatidRolidIndexId)
diff --git a/src/backend/catalog/system_views.sql b/src/backend/catalog/system_views.sql
index 325d4523e6b..2253ca83a74 100644
--- a/src/backend/catalog/system_views.sql
+++ b/src/backend/catalog/system_views.sql
@@ -283,7 +283,37 @@ FROM
 	pg_seclabel l
 	JOIN pg_namespace nsp ON l.classoid = nsp.tableoid AND l.objoid = nsp.oid
 WHERE
-	l.objsubid = 0;
+	l.objsubid = 0
+UNION ALL
+SELECT
+	l.objoid, l.classoid, 0::int4 AS objsubid,
+	'database'::text AS objtype,
+	NULL::oid AS objnamespace,
+	quote_ident(dat.datname) AS objname,
+	l.provider, l.label
+FROM
+	pg_shseclabel l
+	JOIN pg_database dat ON l.classoid = dat.tableoid AND l.objoid = dat.oid
+UNION ALL
+SELECT
+	l.objoid, l.classoid, 0::int4 AS objsubid,
+	'tablespace'::text AS objtype,
+	NULL::oid AS objnamespace,
+	quote_ident(spc.spcname) AS objname,
+	l.provider, l.label
+FROM
+	pg_shseclabel l
+	JOIN pg_tablespace spc ON l.classoid = spc.tableoid AND l.objoid = spc.oid
+UNION ALL
+SELECT
+	l.objoid, l.classoid, 0::int4 AS objsubid,
+	'role'::text AS objtype,
+	NULL::oid AS objnamespace,
+	quote_ident(rol.rolname) AS objname,
+	l.provider, l.label
+FROM
+	pg_shseclabel l
+	JOIN pg_authid rol ON l.classoid = rol.tableoid AND l.objoid = rol.oid;
 
 CREATE VIEW pg_settings AS
     SELECT * FROM pg_show_all_settings() AS A;
diff --git a/src/backend/commands/dbcommands.c b/src/backend/commands/dbcommands.c
index f319eb539c3..93240efbd71 100644
--- a/src/backend/commands/dbcommands.c
+++ b/src/backend/commands/dbcommands.c
@@ -39,6 +39,7 @@
 #include "catalog/pg_tablespace.h"
 #include "commands/comment.h"
 #include "commands/dbcommands.h"
+#include "commands/seclabel.h"
 #include "commands/tablespace.h"
 #include "mb/pg_wchar.h"
 #include "miscadmin.h"
@@ -822,9 +823,11 @@ dropdb(const char *dbname, bool missing_ok)
 	ReleaseSysCache(tup);
 
 	/*
-	 * Delete any comments associated with the database.
+	 * Delete any comments or security labels associated with
+	 * the database.
 	 */
 	DeleteSharedComments(db_id, DatabaseRelationId);
+	DeleteSharedSecurityLabel(db_id, DatabaseRelationId);
 
 	/*
 	 * Remove settings associated with this database
diff --git a/src/backend/commands/seclabel.c b/src/backend/commands/seclabel.c
index 6e7e9c29761..0041734b62b 100644
--- a/src/backend/commands/seclabel.c
+++ b/src/backend/commands/seclabel.c
@@ -16,6 +16,7 @@
 #include "catalog/indexing.h"
 #include "catalog/namespace.h"
 #include "catalog/pg_seclabel.h"
+#include "catalog/pg_shseclabel.h"
 #include "commands/seclabel.h"
 #include "miscadmin.h"
 #include "utils/acl.h"
@@ -24,6 +25,7 @@
 #include "utils/lsyscache.h"
 #include "utils/memutils.h"
 #include "utils/rel.h"
+#include "utils/syscache.h"
 #include "utils/tqual.h"
 
 typedef struct
@@ -135,8 +137,56 @@ ExecSecLabelStmt(SecLabelStmt *stmt)
 }
 
 /*
- * GetSecurityLabel returns the security label for a database object for a
- * given provider, or NULL if there is no such label.
+ * GetSharedSecurityLabel returns the security label for a shared object for
+ * a given provider, or NULL if there is no such label.
+ */
+static char *
+GetSharedSecurityLabel(const ObjectAddress *object, const char *provider)
+{
+	Relation	pg_shseclabel;
+	ScanKeyData keys[3];
+	SysScanDesc scan;
+	HeapTuple	tuple;
+	Datum		datum;
+	bool		isnull;
+	char	   *seclabel = NULL;
+
+	ScanKeyInit(&keys[0],
+				Anum_pg_shseclabel_objoid,
+				BTEqualStrategyNumber, F_OIDEQ,
+				ObjectIdGetDatum(object->objectId));
+	ScanKeyInit(&keys[1],
+				Anum_pg_shseclabel_classoid,
+				BTEqualStrategyNumber, F_OIDEQ,
+				ObjectIdGetDatum(object->classId));
+	ScanKeyInit(&keys[2],
+				Anum_pg_shseclabel_provider,
+				BTEqualStrategyNumber, F_TEXTEQ,
+				CStringGetTextDatum(provider));
+
+	pg_shseclabel = heap_open(SharedSecLabelRelationId, AccessShareLock);
+
+	scan = systable_beginscan(pg_shseclabel, SharedSecLabelObjectIndexId, true,
+							  SnapshotNow, 3, keys);
+
+	tuple = systable_getnext(scan);
+	if (HeapTupleIsValid(tuple))
+	{
+		datum = heap_getattr(tuple, Anum_pg_shseclabel_label,
+							 RelationGetDescr(pg_shseclabel), &isnull);
+		if (!isnull)
+			seclabel = TextDatumGetCString(datum);
+	}
+	systable_endscan(scan);
+
+	heap_close(pg_shseclabel, AccessShareLock);
+
+	return seclabel;
+}
+
+/*
+ * GetSecurityLabel returns the security label for a shared or database object
+ * for a given provider, or NULL if there is no such label.
  */
 char *
 GetSecurityLabel(const ObjectAddress *object, const char *provider)
@@ -149,8 +199,11 @@ GetSecurityLabel(const ObjectAddress *object, const char *provider)
 	bool		isnull;
 	char	   *seclabel = NULL;
 
-	Assert(!IsSharedRelation(object->classId));
+	/* Shared objects have their own security label catalog. */
+	if (IsSharedRelation(object->classId))
+		return GetSharedSecurityLabel(object, provider);
 
+	/* Must be an unshared object, so examine pg_seclabel. */
 	ScanKeyInit(&keys[0],
 				Anum_pg_seclabel_objoid,
 				BTEqualStrategyNumber, F_OIDEQ,
@@ -188,6 +241,84 @@ GetSecurityLabel(const ObjectAddress *object, const char *provider)
 	return seclabel;
 }
 
+/* 
+ * SetSharedSecurityLabel is a helper function of SetSecurityLabel to
+ * handle shared database objects.
+ */
+static void
+SetSharedSecurityLabel(const ObjectAddress *object,
+					   const char *provider, const char *label)
+{
+	Relation	pg_shseclabel;
+	ScanKeyData	keys[4];
+	SysScanDesc	scan;
+	HeapTuple	oldtup;
+	HeapTuple	newtup = NULL;
+	Datum		values[Natts_pg_shseclabel];
+	bool		nulls[Natts_pg_shseclabel];
+	bool		replaces[Natts_pg_shseclabel];
+
+	/* Prepare to form or update a tuple, if necessary. */
+	memset(nulls, false, sizeof(nulls));
+	memset(replaces, false, sizeof(replaces));
+	values[Anum_pg_shseclabel_objoid - 1] = ObjectIdGetDatum(object->objectId);
+	values[Anum_pg_shseclabel_classoid - 1] = ObjectIdGetDatum(object->classId);
+	values[Anum_pg_shseclabel_provider - 1] = CStringGetTextDatum(provider);
+	if (label != NULL)
+		values[Anum_pg_shseclabel_label - 1] = CStringGetTextDatum(label);
+
+	/* Use the index to search for a matching old tuple */
+	ScanKeyInit(&keys[0],
+				Anum_pg_shseclabel_objoid,
+				BTEqualStrategyNumber, F_OIDEQ,
+				ObjectIdGetDatum(object->objectId));
+	ScanKeyInit(&keys[1],
+				Anum_pg_shseclabel_classoid,
+				BTEqualStrategyNumber, F_OIDEQ,
+				ObjectIdGetDatum(object->classId));
+	ScanKeyInit(&keys[2],
+				Anum_pg_shseclabel_provider,
+				BTEqualStrategyNumber, F_TEXTEQ,
+				CStringGetTextDatum(provider));
+
+	pg_shseclabel = heap_open(SharedSecLabelRelationId, RowExclusiveLock);
+
+	scan = systable_beginscan(pg_shseclabel, SharedSecLabelObjectIndexId, true,
+							  SnapshotNow, 3, keys);
+
+	oldtup = systable_getnext(scan);
+	if (HeapTupleIsValid(oldtup))
+	{
+		if (label == NULL)
+			simple_heap_delete(pg_shseclabel, &oldtup->t_self);
+		else
+		{
+			replaces[Anum_pg_shseclabel_label - 1] = true;
+			newtup = heap_modify_tuple(oldtup, RelationGetDescr(pg_shseclabel),
+									   values, nulls, replaces);
+			simple_heap_update(pg_shseclabel, &oldtup->t_self, newtup);
+		}
+	}
+	systable_endscan(scan);
+
+	/* If we didn't find an old tuple, insert a new one */
+	if (newtup == NULL && label != NULL)
+	{
+		newtup = heap_form_tuple(RelationGetDescr(pg_shseclabel),
+								 values, nulls);
+		simple_heap_insert(pg_shseclabel, newtup);
+	}
+
+	/* Update indexes, if necessary */
+	if (newtup != NULL)
+	{
+		CatalogUpdateIndexes(pg_shseclabel, newtup);
+		heap_freetuple(newtup);
+	}
+
+	heap_close(pg_shseclabel, RowExclusiveLock);
+}
+
 /*
  * SetSecurityLabel attempts to set the security label for the specified
  * provider on the specified object to the given value.  NULL means that any
@@ -206,8 +337,12 @@ SetSecurityLabel(const ObjectAddress *object,
 	bool		nulls[Natts_pg_seclabel];
 	bool		replaces[Natts_pg_seclabel];
 
-	/* Security labels on shared objects are not supported. */
-	Assert(!IsSharedRelation(object->classId));
+	/* Shared objects have their own security label catalog. */
+	if (IsSharedRelation(object->classId))
+	{
+		SetSharedSecurityLabel(object, provider, label);
+		return;
+	}
 
 	/* Prepare to form or update a tuple, if necessary. */
 	memset(nulls, false, sizeof(nulls));
@@ -276,6 +411,38 @@ SetSecurityLabel(const ObjectAddress *object,
 }
 
 /*
+ * DeleteSharedSecurityLabel is a helper function of DeleteSecurityLabel
+ * to handle shared database objects.
+ */
+void
+DeleteSharedSecurityLabel(Oid objectId, Oid classId)
+{
+	Relation	pg_shseclabel;
+	ScanKeyData	skey[2];
+	SysScanDesc	scan;
+	HeapTuple	oldtup;
+
+	ScanKeyInit(&skey[0],
+				Anum_pg_shseclabel_objoid,
+				BTEqualStrategyNumber, F_OIDEQ,
+				ObjectIdGetDatum(objectId));
+	ScanKeyInit(&skey[1],
+				Anum_pg_shseclabel_classoid,
+				BTEqualStrategyNumber, F_OIDEQ,
+				ObjectIdGetDatum(classId));
+
+	pg_shseclabel = heap_open(SharedSecLabelRelationId, RowExclusiveLock);
+
+	scan = systable_beginscan(pg_shseclabel, SharedSecLabelObjectIndexId, true,
+							  SnapshotNow, 2, skey);
+	while (HeapTupleIsValid(oldtup = systable_getnext(scan)))
+		simple_heap_delete(pg_shseclabel, &oldtup->t_self);
+	systable_endscan(scan);
+
+	heap_close(pg_shseclabel, RowExclusiveLock);
+}
+
+/*
  * DeleteSecurityLabel removes all security labels for an object (and any
  * sub-objects, if applicable).
  */
@@ -288,9 +455,13 @@ DeleteSecurityLabel(const ObjectAddress *object)
 	HeapTuple	oldtup;
 	int			nkeys;
 
-	/* Security labels on shared objects are not supported. */
+	/* Shared objects have their own security label catalog. */
 	if (IsSharedRelation(object->classId))
+	{
+		Assert(object->objectSubId == 0);
+		DeleteSharedSecurityLabel(object->objectId, object->classId);
 		return;
+	}
 
 	ScanKeyInit(&skey[0],
 				Anum_pg_seclabel_objoid,
diff --git a/src/backend/commands/tablespace.c b/src/backend/commands/tablespace.c
index 3024dc4b646..09ecabb7723 100644
--- a/src/backend/commands/tablespace.c
+++ b/src/backend/commands/tablespace.c
@@ -63,6 +63,7 @@
 #include "catalog/pg_tablespace.h"
 #include "commands/comment.h"
 #include "commands/defrem.h"
+#include "commands/seclabel.h"
 #include "commands/tablespace.h"
 #include "miscadmin.h"
 #include "postmaster/bgwriter.h"
@@ -448,9 +449,10 @@ DropTableSpace(DropTableSpaceStmt *stmt)
 	heap_endscan(scandesc);
 
 	/*
-	 * Remove any comments on this tablespace.
+	 * Remove any comments or security labels on this tablespace.
 	 */
 	DeleteSharedComments(tablespaceoid, TableSpaceRelationId);
+	DeleteSharedSecurityLabel(tablespaceoid, TableSpaceRelationId);
 
 	/*
 	 * Remove dependency on owner.
diff --git a/src/backend/commands/user.c b/src/backend/commands/user.c
index 871bda7cc59..0367b200fef 100644
--- a/src/backend/commands/user.c
+++ b/src/backend/commands/user.c
@@ -24,6 +24,7 @@
 #include "catalog/pg_db_role_setting.h"
 #include "commands/comment.h"
 #include "commands/dbcommands.h"
+#include "commands/seclabel.h"
 #include "commands/user.h"
 #include "libpq/md5.h"
 #include "miscadmin.h"
@@ -1000,9 +1001,10 @@ DropRole(DropRoleStmt *stmt)
 		systable_endscan(sscan);
 
 		/*
-		 * Remove any comments on this role.
+		 * Remove any comments or security labels on this role.
 		 */
 		DeleteSharedComments(roleid, AuthIdRelationId);
+		DeleteSharedSecurityLabel(roleid, AuthIdRelationId);
 
 		/*
 		 * Remove settings for this role.
diff --git a/src/backend/parser/gram.y b/src/backend/parser/gram.y
index 72260320c38..ac094aa5f3a 100644
--- a/src/backend/parser/gram.y
+++ b/src/backend/parser/gram.y
@@ -5068,11 +5068,14 @@ opt_provider:	FOR ColId_or_Sconst	{ $$ = $2; }
 
 security_label_type:
 			COLUMN								{ $$ = OBJECT_COLUMN; }
+			| DATABASE							{ $$ = OBJECT_DATABASE; }
 			| FOREIGN TABLE						{ $$ = OBJECT_FOREIGN_TABLE; }
 			| SCHEMA							{ $$ = OBJECT_SCHEMA; }
 			| SEQUENCE							{ $$ = OBJECT_SEQUENCE; }
 			| TABLE								{ $$ = OBJECT_TABLE; }
 			| DOMAIN_P							{ $$ = OBJECT_TYPE; }
+			| ROLE								{ $$ = OBJECT_ROLE; }
+			| TABLESPACE						{ $$ = OBJECT_TABLESPACE; }
 			| TYPE_P							{ $$ = OBJECT_TYPE; }
 			| VIEW								{ $$ = OBJECT_VIEW; }
 		;