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.

1 files changed, 177 insertions, 6 deletions

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,