Add a SECURITY LABEL command.

This is intended as infrastructure to support integration with label-based
mandatory access control systems such as SE-Linux. Further changes (mostly
hooks) will be needed, but this is a big chunk of it.

KaiGai Kohei and Robert Haas

9 files changed, 641 insertions, 7 deletions

diff --git a/src/backend/catalog/Makefile b/src/backend/catalog/Makefile
index f4a7eb09dca..6a47f398ed8 100644
--- a/src/backend/catalog/Makefile
+++ b/src/backend/catalog/Makefile
@@ -38,7 +38,7 @@ POSTGRES_BKI_SRCS = $(addprefix $(top_srcdir)/src/include/catalog/,\
 	pg_ts_config.h pg_ts_config_map.h pg_ts_dict.h \
 	pg_ts_parser.h pg_ts_template.h \
 	pg_foreign_data_wrapper.h pg_foreign_server.h pg_user_mapping.h \
-	pg_default_acl.h \
+	pg_default_acl.h pg_seclabel.h \
 	toasting.h indexing.h \
     )
 
diff --git a/src/backend/catalog/dependency.c b/src/backend/catalog/dependency.c
index 62598ee8f89..18e07eb956a 100644
--- a/src/backend/catalog/dependency.c
+++ b/src/backend/catalog/dependency.c
@@ -57,6 +57,7 @@
 #include "commands/defrem.h"
 #include "commands/proclang.h"
 #include "commands/schemacmds.h"
+#include "commands/seclabel.h"
 #include "commands/tablespace.h"
 #include "commands/trigger.h"
 #include "commands/typecmds.h"
@@ -1004,10 +1005,12 @@ deleteOneObject(const ObjectAddress *object, Relation depRel)
 	doDeletion(object);
 
 	/*
-	 * Delete any comments associated with this object.  (This is a convenient
-	 * place to do it instead of having every object type know to do it.)
+	 * Delete any comments or security labels associated with this object.
+	 * (This is a convenient place to do these things, rather than having every
+	 * object type know to do it.)
 	 */
 	DeleteComments(object->objectId, object->classId, object->objectSubId);
+	DeleteSecurityLabel(object);
 
 	/*
 	 * CommandCounterIncrement here to ensure that preceding changes are all
diff --git a/src/backend/catalog/system_views.sql b/src/backend/catalog/system_views.sql
index 651ffc61b96..09574c3e82c 100644
--- a/src/backend/catalog/system_views.sql
+++ b/src/backend/catalog/system_views.sql
@@ -160,6 +160,114 @@ CREATE VIEW pg_prepared_xacts AS
 CREATE VIEW pg_prepared_statements AS
     SELECT * FROM pg_prepared_statement() AS P;
 
+CREATE VIEW pg_seclabels AS
+SELECT
+	l.objoid, l.classoid, l.objsubid,
+	CASE WHEN rel.relkind = 'r' THEN 'table'::text
+	     WHEN rel.relkind = 'v' THEN 'view'::text
+	     WHEN rel.relkind = 'S' THEN 'sequence'::text END AS objtype,
+	rel.relnamespace AS objnamespace,
+	CASE WHEN pg_table_is_visible(rel.oid)
+	     THEN quote_ident(rel.relname)
+	     ELSE quote_ident(nsp.nspname) || '.' || quote_ident(rel.relname)
+	     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)
+	     ELSE quote_ident(nsp.nspname) || '.' || quote_ident(rel.relname)
+	     END || '.' || att.attname 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
+	END AS objtype,
+	pro.pronamespace AS objnamespace,
+	CASE WHEN pg_function_is_visible(pro.oid)
+	     THEN quote_ident(pro.proname)
+	     ELSE quote_ident(nsp.nspname) || '.' || quote_ident(pro.proname)
+	END || '(' || pg_catalog.pg_get_function_arguments(pro.oid) || ')' 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' 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)
+	ELSE quote_ident(nsp.nspname) || '.' || quote_ident(typ.typname)
+	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_catalog.pg_largeobject'::regclass 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) 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) 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;
+
 CREATE VIEW pg_settings AS 
     SELECT * FROM pg_show_all_settings() AS A; 
 
diff --git a/src/backend/commands/Makefile b/src/backend/commands/Makefile
index 4e9bf43ad5f..9d2a7322457 100644
--- a/src/backend/commands/Makefile
+++ b/src/backend/commands/Makefile
@@ -17,7 +17,7 @@ OBJS = aggregatecmds.o alter.o analyze.o async.o cluster.o comment.o  \
 	dbcommands.o define.o discard.o explain.o foreigncmds.o functioncmds.o \
 	indexcmds.o lockcmds.o operatorcmds.o opclasscmds.o \
 	portalcmds.o prepare.o proclang.o \
-	schemacmds.o sequence.o tablecmds.o tablespace.o trigger.o \
+	schemacmds.o seclabel.o sequence.o tablecmds.o tablespace.o trigger.o \
 	tsearchcmds.o typecmds.o user.o vacuum.o vacuumlazy.o \
 	variable.o view.o
 
diff --git a/src/backend/commands/seclabel.c b/src/backend/commands/seclabel.c
new file mode 100644
index 00000000000..417ad88d557
--- /dev/null
+++ b/src/backend/commands/seclabel.c
@@ -0,0 +1,387 @@
+/* -------------------------------------------------------------------------
+ *
+ * seclabel.c
+ *    routines to support security label feature.
+ *
+ * Portions Copyright (c) 1996-2010, PostgreSQL Global Development Group
+ * Portions Copyright (c) 1994, Regents of the University of California
+ *
+ * -------------------------------------------------------------------------
+ */
+#include "postgres.h"
+
+#include "access/genam.h"
+#include "access/heapam.h"
+#include "catalog/catalog.h"
+#include "catalog/indexing.h"
+#include "catalog/namespace.h"
+#include "catalog/pg_seclabel.h"
+#include "commands/seclabel.h"
+#include "miscadmin.h"
+#include "utils/acl.h"
+#include "utils/builtins.h"
+#include "utils/fmgroids.h"
+#include "utils/lsyscache.h"
+#include "utils/memutils.h"
+#include "utils/tqual.h"
+
+/*
+ * For most object types, the permissions-checking logic is simple enough
+ * that it makes sense to just include it in CommentObject().  However,
+ * attributes require a bit more checking.
+ */
+static void CheckAttributeSecLabel(Relation relation);
+
+typedef struct
+{
+	const char *provider_name;
+	check_object_relabel_type	hook;
+} LabelProvider;
+
+static List *label_provider_list = NIL;
+
+/*
+ * ExecSecLabelStmt --
+ *
+ * Apply a security label to a database object.
+ */
+void
+ExecSecLabelStmt(SecLabelStmt *stmt)
+{
+	LabelProvider *provider = NULL;
+	ObjectAddress	address;
+	Relation		relation;
+	ListCell	   *lc;
+
+	/*
+	 * Find the named label provider, or if none specified, check whether
+	 * there's exactly one, and if so use it.
+	 */
+	if (stmt->provider == NULL)
+	{
+		if (label_provider_list == NIL)
+			ereport(ERROR,
+					(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
+				 errmsg("security label providers have been loaded")));
+		if (lnext(list_head(label_provider_list)) != NULL)
+			ereport(ERROR,
+					(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
+				 errmsg("must specify provider when multiple security label providers have been loaded")));
+		provider = (LabelProvider *) linitial(label_provider_list);
+	}
+	else
+	{
+		foreach (lc, label_provider_list)
+		{
+			LabelProvider *lp = lfirst(lc);
+
+			if (strcmp(stmt->provider, lp->provider_name) == 0)
+			{
+				provider = lp;
+				break;
+			}
+		}
+		if (provider == NULL)
+			ereport(ERROR,
+					(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
+					 errmsg("security label provider \"%s\" is not loaded",
+							stmt->provider)));
+	}
+
+	/*
+	 * Translate the parser representation which identifies this object
+	 * into an ObjectAddress. get_object_address() will throw an error if
+     * the object does not exist, and will also acquire a lock on the
+     * target to guard against concurrent modifications.
+	 */
+	address = get_object_address(stmt->objtype, stmt->objname, stmt->objargs,
+								 &relation, ShareUpdateExclusiveLock);
+
+	/* Privilege and integrity checks. */
+	switch (stmt->objtype)
+	{
+		case OBJECT_SEQUENCE:
+		case OBJECT_TABLE:
+		case OBJECT_VIEW:
+			if (!pg_class_ownercheck(RelationGetRelid(relation), GetUserId()))
+				aclcheck_error(ACLCHECK_NOT_OWNER, ACL_KIND_CLASS,
+							   RelationGetRelationName(relation));
+			break;
+		case OBJECT_COLUMN:
+			CheckAttributeSecLabel(relation);
+			break;
+		case OBJECT_TYPE:
+			if (!pg_type_ownercheck(address.objectId, GetUserId()))
+				aclcheck_error(ACLCHECK_NOT_OWNER, ACL_KIND_TYPE,
+							   format_type_be(address.objectId));
+			break;
+		case OBJECT_AGGREGATE:
+		case OBJECT_FUNCTION:
+			if (!pg_proc_ownercheck(address.objectId, GetUserId()))
+				aclcheck_error(ACLCHECK_NOT_OWNER, ACL_KIND_PROC,
+							   NameListToString(stmt->objname));
+			break;
+		case OBJECT_SCHEMA:
+			if (!pg_namespace_ownercheck(address.objectId, GetUserId()))
+				aclcheck_error(ACLCHECK_NOT_OWNER, ACL_KIND_NAMESPACE,
+							   strVal(linitial(stmt->objname)));
+			break;
+		case OBJECT_LANGUAGE:
+			if (!superuser())
+				ereport(ERROR,
+						(errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
+					 errmsg("must be superuser to comment on procedural language")));
+			break;
+		case OBJECT_LARGEOBJECT:
+			if (!pg_largeobject_ownercheck(address.objectId, GetUserId()))
+				ereport(ERROR,
+						(errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
+						 errmsg("must be owner of large object %u",
+							address.objectId)));
+			break;
+		default:
+			elog(ERROR, "unrecognized object type: %d",
+				 (int) stmt->objtype);
+	}
+
+	/* Provider gets control here, may throw ERROR to veto new label. */
+	(*provider->hook)(&address, stmt->label);
+
+	/* Apply new label. */
+	SetSecurityLabel(&address, provider->provider_name, stmt->label);
+
+	/*
+	 * If get_object_address() opened the relation for us, we close it to keep
+	 * the reference count correct - but we retain any locks acquired by
+	 * get_object_address() until commit time, to guard against concurrent
+	 * activity.
+	 */
+	if (relation != NULL)
+		relation_close(relation, NoLock);
+}
+
+/*
+ * GetSecurityLabel returns the security label for a database object for a
+ * given provider, or NULL if there is no such label.
+ */
+char *
+GetSecurityLabel(const ObjectAddress *object, const char *provider)
+{
+	Relation	pg_seclabel;
+	ScanKeyData	keys[4];
+	SysScanDesc	scan;
+	HeapTuple	tuple;
+	Datum		datum;
+	bool		isnull;
+	char	   *seclabel = NULL;
+
+	Assert(!IsSharedRelation(object->classId));
+
+	ScanKeyInit(&keys[0],
+				Anum_pg_seclabel_objoid,
+				BTEqualStrategyNumber, F_OIDEQ,
+				ObjectIdGetDatum(object->objectId));
+	ScanKeyInit(&keys[1],
+				Anum_pg_seclabel_classoid,
+				BTEqualStrategyNumber, F_OIDEQ,
+				ObjectIdGetDatum(object->classId));
+	ScanKeyInit(&keys[2],
+				Anum_pg_seclabel_objsubid,
+				BTEqualStrategyNumber, F_INT4EQ,
+				Int32GetDatum(object->objectSubId));
+	ScanKeyInit(&keys[3],
+				Anum_pg_seclabel_provider,
+				BTEqualStrategyNumber, F_TEXTEQ,
+				CStringGetTextDatum(provider));
+
+	pg_seclabel = heap_open(SecLabelRelationId, AccessShareLock);
+
+	scan = systable_beginscan(pg_seclabel, SecLabelObjectIndexId, true,
+							  SnapshotNow, 4, keys);
+
+	tuple = systable_getnext(scan);
+	if (HeapTupleIsValid(tuple))
+	{
+		datum = heap_getattr(tuple, Anum_pg_seclabel_label,
+							 RelationGetDescr(pg_seclabel), &isnull);
+		if (!isnull)
+			seclabel = TextDatumGetCString(datum);
+	}
+	systable_endscan(scan);
+
+	heap_close(pg_seclabel, AccessShareLock);
+
+	return seclabel;
+}
+
+/*
+ * SetSecurityLabel attempts to set the security label for the specified
+ * provider on the specified object to the given value.  NULL means that any
+ * any existing label should be deleted.
+ */
+void
+SetSecurityLabel(const ObjectAddress *object,
+				 const char *provider, const char *label)
+{
+	Relation	pg_seclabel;
+	ScanKeyData	keys[4];
+	SysScanDesc	scan;
+	HeapTuple	oldtup;
+	HeapTuple	newtup = NULL;
+	Datum		values[Natts_pg_seclabel];
+	bool		nulls[Natts_pg_seclabel];
+	bool		replaces[Natts_pg_seclabel];
+
+	/* Security labels on shared objects are not supported. */
+	Assert(!IsSharedRelation(object->classId));
+
+	/* Prepare to form or update a tuple, if necessary. */
+	memset(nulls, false, sizeof(nulls));
+	memset(replaces, false, sizeof(replaces));
+	values[Anum_pg_seclabel_objoid - 1] = ObjectIdGetDatum(object->objectId);
+	values[Anum_pg_seclabel_classoid - 1] = ObjectIdGetDatum(object->classId);
+	values[Anum_pg_seclabel_objsubid - 1] = Int32GetDatum(object->objectSubId);
+	values[Anum_pg_seclabel_provider - 1] = CStringGetTextDatum(provider);
+	if (label != NULL)
+		values[Anum_pg_seclabel_label - 1] = CStringGetTextDatum(label);
+
+	/* Use the index to search for a matching old tuple */
+	ScanKeyInit(&keys[0],
+				Anum_pg_seclabel_objoid,
+				BTEqualStrategyNumber, F_OIDEQ,
+				ObjectIdGetDatum(object->objectId));
+	ScanKeyInit(&keys[1],
+				Anum_pg_seclabel_classoid,
+				BTEqualStrategyNumber, F_OIDEQ,
+				ObjectIdGetDatum(object->classId));
+	ScanKeyInit(&keys[2],
+				Anum_pg_seclabel_objsubid,
+				BTEqualStrategyNumber, F_INT4EQ,
+				Int32GetDatum(object->objectSubId));
+	ScanKeyInit(&keys[3],
+				Anum_pg_seclabel_provider,
+				BTEqualStrategyNumber, F_TEXTEQ,
+				CStringGetTextDatum(provider));
+
+	pg_seclabel = heap_open(SecLabelRelationId, RowExclusiveLock);
+
+	scan = systable_beginscan(pg_seclabel, SecLabelObjectIndexId, true,
+							  SnapshotNow, 4, keys);
+
+	oldtup = systable_getnext(scan);
+	if (HeapTupleIsValid(oldtup))
+	{
+		if (label == NULL)
+			simple_heap_delete(pg_seclabel, &oldtup->t_self);
+		else
+		{
+			replaces[Anum_pg_seclabel_label - 1] = true;
+			newtup = heap_modify_tuple(oldtup, RelationGetDescr(pg_seclabel),
+									   values, nulls, replaces);
+			simple_heap_update(pg_seclabel, &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_seclabel),
+								 values, nulls);
+		simple_heap_insert(pg_seclabel, newtup);
+	}
+
+	/* Update indexes, if necessary */
+	if (newtup != NULL)
+	{
+		CatalogUpdateIndexes(pg_seclabel, newtup);
+		heap_freetuple(newtup);
+	}
+
+	heap_close(pg_seclabel, RowExclusiveLock);
+}
+
+/*
+ * DeleteSecurityLabel removes all security labels for an object (and any
+ * sub-objects, if applicable).
+ */
+void
+DeleteSecurityLabel(const ObjectAddress *object)
+{
+	Relation	pg_seclabel;
+	ScanKeyData	skey[3];
+	SysScanDesc	scan;
+	HeapTuple	oldtup;
+	int			nkeys;
+
+	/* Security labels on shared objects are not supported. */
+	if (IsSharedRelation(object->classId))
+		return;
+
+	ScanKeyInit(&skey[0],
+				Anum_pg_seclabel_objoid,
+				BTEqualStrategyNumber, F_OIDEQ,
+				ObjectIdGetDatum(object->objectId));
+	ScanKeyInit(&skey[1],
+				Anum_pg_seclabel_classoid,
+				BTEqualStrategyNumber, F_OIDEQ,
+				ObjectIdGetDatum(object->classId));
+	if (object->objectSubId != 0)
+	{
+		ScanKeyInit(&skey[2],
+					Anum_pg_seclabel_objsubid,
+					BTEqualStrategyNumber, F_INT4EQ,
+					Int32GetDatum(object->objectSubId));
+		nkeys = 3;
+	}
+	else
+		nkeys = 2;
+
+	pg_seclabel = heap_open(SecLabelRelationId, RowExclusiveLock);
+
+	scan = systable_beginscan(pg_seclabel, SecLabelObjectIndexId, true,
+							  SnapshotNow, nkeys, skey);
+	while (HeapTupleIsValid(oldtup = systable_getnext(scan)))
+		simple_heap_delete(pg_seclabel, &oldtup->t_self);
+	systable_endscan(scan);
+
+	heap_close(pg_seclabel, RowExclusiveLock);
+}
+
+/*
+ * Check whether the user is allowed to comment on an attribute of the
+ * specified relation.
+ */
+static void
+CheckAttributeSecLabel(Relation relation)
+{
+	if (!pg_class_ownercheck(RelationGetRelid(relation), GetUserId()))
+		aclcheck_error(ACLCHECK_NOT_OWNER, ACL_KIND_CLASS,
+					   RelationGetRelationName(relation));
+
+	/*
+	 * Allow security labels only on columns of tables, views, and composite
+	 * types (which are the only relkinds for which pg_dump will dump labels).
+	 */
+	if (relation->rd_rel->relkind != RELKIND_RELATION &&
+		relation->rd_rel->relkind != RELKIND_VIEW &&
+		relation->rd_rel->relkind != RELKIND_COMPOSITE_TYPE)
+		ereport(ERROR,
+				(errcode(ERRCODE_WRONG_OBJECT_TYPE),
+				 errmsg("\"%s\" is not a table, view, or composite type",
+						RelationGetRelationName(relation))));
+}
+
+void
+register_label_provider(const char *provider_name, check_object_relabel_type hook)
+{
+	LabelProvider  *provider;
+	MemoryContext	oldcxt;
+
+	oldcxt = MemoryContextSwitchTo(TopMemoryContext);
+	provider = palloc(sizeof(LabelProvider));
+	provider->provider_name = pstrdup(provider_name);
+	provider->hook = hook;
+	label_provider_list = lappend(label_provider_list, provider);
+	MemoryContextSwitchTo(oldcxt);
+}
diff --git a/src/backend/nodes/copyfuncs.c b/src/backend/nodes/copyfuncs.c
index deaeb761d4a..e07aa3ead23 100644
--- a/src/backend/nodes/copyfuncs.c
+++ b/src/backend/nodes/copyfuncs.c
@@ -2607,6 +2607,20 @@ _copyCommentStmt(CommentStmt *from)
 	return newnode;
 }
 
+static SecLabelStmt *
+_copySecLabelStmt(SecLabelStmt *from)
+{
+	SecLabelStmt *newnode = makeNode(SecLabelStmt);
+
+	COPY_SCALAR_FIELD(objtype);
+	COPY_NODE_FIELD(objname);
+	COPY_NODE_FIELD(objargs);
+	COPY_STRING_FIELD(provider);
+	COPY_STRING_FIELD(label);
+
+	return newnode;
+}
+
 static FetchStmt *
 _copyFetchStmt(FetchStmt *from)
 {
@@ -3958,6 +3972,9 @@ copyObject(void *from)
 		case T_CommentStmt:
 			retval = _copyCommentStmt(from);
 			break;
+		case T_SecLabelStmt:
+			retval = _copySecLabelStmt(from);
+			break;
 		case T_FetchStmt:
 			retval = _copyFetchStmt(from);
 			break;
diff --git a/src/backend/nodes/equalfuncs.c b/src/backend/nodes/equalfuncs.c
index 6b6cd9966ce..8d083c8796d 100644
--- a/src/backend/nodes/equalfuncs.c
+++ b/src/backend/nodes/equalfuncs.c
@@ -1164,6 +1164,18 @@ _equalCommentStmt(CommentStmt *a, CommentStmt *b)
 }
 
 static bool
+_equalSecLabelStmt(SecLabelStmt *a, SecLabelStmt *b)
+{
+	COMPARE_SCALAR_FIELD(objtype);
+	COMPARE_NODE_FIELD(objname);
+	COMPARE_NODE_FIELD(objargs);
+	COMPARE_STRING_FIELD(provider);
+	COMPARE_STRING_FIELD(label);
+
+	return true;
+}
+
+static bool
 _equalFetchStmt(FetchStmt *a, FetchStmt *b)
 {
 	COMPARE_SCALAR_FIELD(direction);
@@ -2624,6 +2636,9 @@ equal(void *a, void *b)
 		case T_CommentStmt:
 			retval = _equalCommentStmt(a, b);
 			break;
+		case T_SecLabelStmt:
+			retval = _equalSecLabelStmt(a, b);
+			break;
 		case T_FetchStmt:
 			retval = _equalFetchStmt(a, b);
 			break;
diff --git a/src/backend/parser/gram.y b/src/backend/parser/gram.y
index 40bd7a39325..4054cb1bc7b 100644
--- a/src/backend/parser/gram.y
+++ b/src/backend/parser/gram.y
@@ -205,7 +205,7 @@ static RangeVar *makeRangeVarFromAnyName(List *names, int position, core_yyscan_
 		CreateFunctionStmt AlterFunctionStmt ReindexStmt RemoveAggrStmt
 		RemoveFuncStmt RemoveOperStmt RenameStmt RevokeStmt RevokeRoleStmt
 		RuleActionStmt RuleActionStmtOrEmpty RuleStmt
-		SelectStmt TransactionStmt TruncateStmt
+		SecLabelStmt SelectStmt TransactionStmt TruncateStmt
 		UnlistenStmt UpdateStmt VacuumStmt
 		VariableResetStmt VariableSetStmt VariableShowStmt
 		ViewStmt CheckPointStmt CreateConversionStmt
@@ -335,7 +335,7 @@ static RangeVar *makeRangeVarFromAnyName(List *names, int position, core_yyscan_
 %type <boolean> copy_from
 
 %type <ival>	opt_column event cursor_options opt_hold opt_set_data
-%type <objtype>	reindex_type drop_type comment_type
+%type <objtype>	reindex_type drop_type comment_type security_label_type
 
 %type <node>	fetch_args limit_clause select_limit_value
 				offset_clause select_offset_value
@@ -423,6 +423,8 @@ static RangeVar *makeRangeVarFromAnyName(List *names, int position, core_yyscan_
 %type <str>		OptTableSpace OptConsTableSpace OptTableSpaceOwner
 %type <list>	opt_check_option
 
+%type <str>		opt_provider security_label
+
 %type <target>	xml_attribute_el
 %type <list>	xml_attribute_list xml_attributes
 %type <node>	xml_root_version opt_xml_root_standalone
@@ -500,7 +502,7 @@ static RangeVar *makeRangeVarFromAnyName(List *names, int position, core_yyscan_
 
 	KEY
 
-	LANGUAGE LARGE_P LAST_P LC_COLLATE_P LC_CTYPE_P LEADING
+	LABEL LANGUAGE LARGE_P LAST_P LC_COLLATE_P LC_CTYPE_P LEADING
 	LEAST LEFT LEVEL LIKE LIMIT LISTEN LOAD LOCAL LOCALTIME LOCALTIMESTAMP
 	LOCATION LOCK_P LOGIN_P
 
@@ -739,6 +741,7 @@ stmt :
 			| RevokeStmt
 			| RevokeRoleStmt
 			| RuleStmt
+			| SecLabelStmt
 			| SelectStmt
 			| TransactionStmt
 			| TruncateStmt
@@ -4368,6 +4371,92 @@ comment_text:
 			| NULL_P							{ $$ = NULL; }
 		;
 
+
+/*****************************************************************************
+ *
+ *  SECURITY LABEL [FOR <provider>] ON <object> IS <label>
+ *
+ *  As with COMMENT ON, <object> can refer to various types of database
+ *  objects (e.g. TABLE, COLUMN, etc.).
+ *
+ *****************************************************************************/
+
+SecLabelStmt:
+			SECURITY LABEL opt_provider ON security_label_type any_name
+			IS security_label
+				{
+					SecLabelStmt *n = makeNode(SecLabelStmt);
+					n->provider = $3;
+					n->objtype = $5;
+					n->objname = $6;
+					n->objargs = NIL;
+					n->label = $8;
+					$$ = (Node *) n;
+				}
+			| SECURITY LABEL opt_provider ON AGGREGATE func_name aggr_args
+			  IS security_label
+				{
+					SecLabelStmt *n = makeNode(SecLabelStmt);
+					n->provider = $3;
+					n->objtype = OBJECT_AGGREGATE;
+					n->objname = $6;
+					n->objargs = $7;
+					n->label = $9;
+					$$ = (Node *) n;
+				}
+			| SECURITY LABEL opt_provider ON FUNCTION func_name func_args
+			  IS security_label
+				{
+					SecLabelStmt *n = makeNode(SecLabelStmt);
+					n->provider = $3;
+					n->objtype = OBJECT_FUNCTION;
+					n->objname = $6;
+					n->objargs = extractArgTypes($7);
+					n->label = $9;
+					$$ = (Node *) n;
+				}
+			| SECURITY LABEL opt_provider ON LARGE_P OBJECT_P NumericOnly
+			  IS security_label
+				{
+					SecLabelStmt *n = makeNode(SecLabelStmt);
+					n->provider = $3;
+					n->objtype = OBJECT_LARGEOBJECT;
+					n->objname = list_make1($7);
+					n->objargs = NIL;
+					n->label = $9;
+					$$ = (Node *) n;
+				}
+			| SECURITY LABEL opt_provider ON opt_procedural LANGUAGE any_name
+			  IS security_label
+				{
+					SecLabelStmt *n = makeNode(SecLabelStmt);
+					n->provider = $3;
+					n->objtype = OBJECT_LANGUAGE;
+					n->objname = $7;
+					n->objargs = NIL;
+					n->label = $9;
+					$$ = (Node *) n;
+				}
+		;
+
+opt_provider:	FOR ColId_or_Sconst	{ $$ = $2; }
+				| /* empty */		{ $$ = NULL; }
+		;
+
+security_label_type:
+			COLUMN								{ $$ = OBJECT_COLUMN; }
+			| SCHEMA							{ $$ = OBJECT_SCHEMA; }
+			| SEQUENCE							{ $$ = OBJECT_SEQUENCE; }
+			| TABLE								{ $$ = OBJECT_TABLE; }
+			| DOMAIN_P							{ $$ = OBJECT_TYPE; }
+			| TYPE_P							{ $$ = OBJECT_TYPE; }
+			| VIEW								{ $$ = OBJECT_VIEW; }
+		;
+
+security_label:	Sconst				{ $$ = $1; }
+				| NULL_P			{ $$ = NULL; }
+		;
+
 /*****************************************************************************
  *
  *		QUERY:
@@ -11049,6 +11138,7 @@ unreserved_keyword:
 			| INVOKER
 			| ISOLATION
 			| KEY
+			| LABEL
 			| LANGUAGE
 			| LARGE_P
 			| LAST_P
diff --git a/src/backend/tcop/utility.c b/src/backend/tcop/utility.c
index f69559f0e43..75cb354ea89 100644
--- a/src/backend/tcop/utility.c
+++ b/src/backend/tcop/utility.c
@@ -37,6 +37,7 @@
 #include "commands/prepare.h"
 #include "commands/proclang.h"
 #include "commands/schemacmds.h"
+#include "commands/seclabel.h"
 #include "commands/sequence.h"
 #include "commands/tablecmds.h"
 #include "commands/tablespace.h"
@@ -218,6 +219,7 @@ check_xact_readonly(Node *parsetree)
 		case T_AlterUserMappingStmt:
 		case T_DropUserMappingStmt:
 		case T_AlterTableSpaceOptionsStmt:
+		case T_SecLabelStmt:
 			PreventCommandIfReadOnly(CreateCommandTag(parsetree));
 			break;
 		default:
@@ -663,6 +665,10 @@ standard_ProcessUtility(Node *parsetree,
 			CommentObject((CommentStmt *) parsetree);
 			break;
 
+		case T_SecLabelStmt:
+			ExecSecLabelStmt((SecLabelStmt *) parsetree);
+			break;
+
 		case T_CopyStmt:
 			{
 				uint64		processed;
@@ -1592,6 +1598,10 @@ CreateCommandTag(Node *parsetree)
 			tag = "COMMENT";
 			break;
 
+		case T_SecLabelStmt:
+			tag = "SECURITY LABEL";
+			break;
+
 		case T_CopyStmt:
 			tag = "COPY";
 			break;
@@ -2318,6 +2328,10 @@ GetCommandLogLevel(Node *parsetree)
 			lev = LOGSTMT_DDL;
 			break;
 
+		case T_SecLabelStmt:
+			lev = LOGSTMT_DDL;
+			break;
+
 		case T_CopyStmt:
 			if (((CopyStmt *) parsetree)->is_from)
 				lev = LOGSTMT_MOD;