Allow non-superuser database owners to create procedural languages.

A DBA is allowed to create a language in his database if it's marked
"tmpldbacreate" in pg_pltemplate.  The factory default is that this is set
for all standard trusted languages, but of course a superuser may adjust
the settings.  In service of this, add the long-foreseen owner column to
pg_language; renaming, dropping, and altering owner of a PL now follow
normal ownership rules instead of being superuser-only.
Jeremy Drake, with some editorialization by Tom Lane.

5 files changed, 188 insertions, 39 deletions

diff --git a/src/backend/catalog/aclchk.c b/src/backend/catalog/aclchk.c
index a2b64c78b3f..b837ba4e91c 100644
--- a/src/backend/catalog/aclchk.c
+++ b/src/backend/catalog/aclchk.c
@@ -8,7 +8,7 @@
  *
  *
  * IDENTIFICATION
- *	  $PostgreSQL: pgsql/src/backend/catalog/aclchk.c,v 1.137 2007/02/14 01:58:56 tgl Exp $
+ *	  $PostgreSQL: pgsql/src/backend/catalog/aclchk.c,v 1.138 2007/03/26 16:58:38 tgl Exp $
  *
  * NOTES
  *	  See acl.h.
@@ -1003,11 +1003,8 @@ ExecGrant_Language(InternalGrant *istmt)
 		/*
 		 * Get owner ID and working copy of existing ACL. If there's no ACL,
 		 * substitute the proper default.
-		 *
-		 * Note: for now, languages are treated as owned by the bootstrap
-		 * user. We should add an owner column to pg_language instead.
 		 */
-		ownerId = BOOTSTRAP_SUPERUSERID;
+		ownerId = pg_language_tuple->lanowner;
 		aclDatum = SysCacheGetAttr(LANGNAME, tuple, Anum_pg_language_lanacl,
 								   &isNull);
 		if (isNull)
@@ -1770,8 +1767,7 @@ pg_language_aclmask(Oid lang_oid, Oid roleid,
 				(errcode(ERRCODE_UNDEFINED_OBJECT),
 				 errmsg("language with OID %u does not exist", lang_oid)));
 
-	/* XXX pg_language should have an owner column, but doesn't */
-	ownerId = BOOTSTRAP_SUPERUSERID;
+	ownerId = ((Form_pg_language) GETSTRUCT(tuple))->lanowner;
 
 	aclDatum = SysCacheGetAttr(LANGOID, tuple, Anum_pg_language_lanacl,
 							   &isNull);
@@ -2148,6 +2144,34 @@ pg_proc_ownercheck(Oid proc_oid, Oid roleid)
 }
 
 /*
+ * Ownership check for a procedural language (specified by OID)
+ */
+bool
+pg_language_ownercheck(Oid lan_oid, Oid roleid)
+{
+	HeapTuple	tuple;
+	Oid			ownerId;
+
+	/* Superusers bypass all permission checking. */
+	if (superuser_arg(roleid))
+		return true;
+
+	tuple = SearchSysCache(LANGOID,
+						   ObjectIdGetDatum(lan_oid),
+						   0, 0, 0);
+	if (!HeapTupleIsValid(tuple))
+		ereport(ERROR,
+				(errcode(ERRCODE_UNDEFINED_FUNCTION),
+				 errmsg("language with OID %u does not exist", lan_oid)));
+
+	ownerId = ((Form_pg_language) GETSTRUCT(tuple))->lanowner;
+
+	ReleaseSysCache(tuple);
+
+	return has_privs_of_role(roleid, ownerId);
+}
+
+/*
  * Ownership check for a namespace (specified by OID).
  */
 bool
diff --git a/src/backend/commands/alter.c b/src/backend/commands/alter.c
index 30b7ebde002..118e4adb166 100644
--- a/src/backend/commands/alter.c
+++ b/src/backend/commands/alter.c
@@ -8,7 +8,7 @@
  *
  *
  * IDENTIFICATION
- *	  $PostgreSQL: pgsql/src/backend/commands/alter.c,v 1.22 2007/01/23 05:07:17 tgl Exp $
+ *	  $PostgreSQL: pgsql/src/backend/commands/alter.c,v 1.23 2007/03/26 16:58:38 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -203,6 +203,10 @@ ExecAlterOwnerStmt(AlterOwnerStmt *stmt)
 			AlterFunctionOwner(stmt->object, stmt->objarg, newowner);
 			break;
 
+		case OBJECT_LANGUAGE:
+			AlterLanguageOwner((char *) linitial(stmt->object), newowner);
+			break;
+
 		case OBJECT_OPERATOR:
 			Assert(list_length(stmt->objarg) == 2);
 			AlterOperatorOwner(stmt->object,
diff --git a/src/backend/commands/proclang.c b/src/backend/commands/proclang.c
index 028ab5e5217..e396e0018d0 100644
--- a/src/backend/commands/proclang.c
+++ b/src/backend/commands/proclang.c
@@ -7,7 +7,7 @@
  * Portions Copyright (c) 1994, Regents of the University of California
  *
  * IDENTIFICATION
- *	  $PostgreSQL: pgsql/src/backend/commands/proclang.c,v 1.71 2007/01/22 01:35:20 tgl Exp $
+ *	  $PostgreSQL: pgsql/src/backend/commands/proclang.c,v 1.72 2007/03/26 16:58:38 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -17,16 +17,19 @@
 #include "access/heapam.h"
 #include "catalog/dependency.h"
 #include "catalog/indexing.h"
+#include "catalog/pg_authid.h"
 #include "catalog/pg_language.h"
 #include "catalog/pg_namespace.h"
 #include "catalog/pg_pltemplate.h"
 #include "catalog/pg_proc.h"
 #include "catalog/pg_type.h"
+#include "commands/dbcommands.h"
 #include "commands/defrem.h"
 #include "commands/proclang.h"
 #include "miscadmin.h"
 #include "parser/gramparse.h"
 #include "parser/parse_func.h"
+#include "utils/acl.h"
 #include "utils/builtins.h"
 #include "utils/fmgroids.h"
 #include "utils/lsyscache.h"
@@ -36,13 +39,14 @@
 typedef struct
 {
 	bool		tmpltrusted;	/* trusted? */
+	bool		tmpldbacreate;	/* db owner allowed to create? */
 	char	   *tmplhandler;	/* name of handler function */
 	char	   *tmplvalidator;	/* name of validator function, or NULL */
 	char	   *tmpllibrary;	/* path of shared library */
 } PLTemplate;
 
 static void create_proc_lang(const char *languageName,
-				 Oid handlerOid, Oid valOid, bool trusted);
+				 Oid languageOwner, Oid handlerOid, Oid valOid, bool trusted);
 static PLTemplate *find_language_template(const char *languageName);
 
 
@@ -61,14 +65,6 @@ CreateProceduralLanguage(CreatePLangStmt *stmt)
 	Oid			funcargtypes[1];
 
 	/*
-	 * Check permission
-	 */
-	if (!superuser())
-		ereport(ERROR,
-				(errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
-				 errmsg("must be superuser to create procedural language")));
-
-	/*
 	 * Translate the language name and check that this language doesn't
 	 * already exist
 	 */
@@ -97,6 +93,21 @@ CreateProceduralLanguage(CreatePLangStmt *stmt)
 					(errmsg("using pg_pltemplate information instead of CREATE LANGUAGE parameters")));
 
 		/*
+		 * Check permission
+		 */
+		if (!superuser())
+		{
+			if (!pltemplate->tmpldbacreate)
+				ereport(ERROR,
+						(errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
+						 errmsg("must be superuser to create procedural language \"%s\"",
+								languageName)));
+			if (!pg_database_ownercheck(MyDatabaseId, GetUserId()))
+				aclcheck_error(ACLCHECK_NOT_OWNER, ACL_KIND_DATABASE,
+							   get_database_name(MyDatabaseId));
+		}
+
+		/*
 		 * Find or create the handler function, which we force to be in the
 		 * pg_catalog schema.  If already present, it must have the correct
 		 * return type.
@@ -171,7 +182,7 @@ CreateProceduralLanguage(CreatePLangStmt *stmt)
 			valOid = InvalidOid;
 
 		/* ok, create it */
-		create_proc_lang(languageName, handlerOid, valOid,
+		create_proc_lang(languageName, GetUserId(), handlerOid, valOid,
 						 pltemplate->tmpltrusted);
 	}
 	else
@@ -189,6 +200,14 @@ CreateProceduralLanguage(CreatePLangStmt *stmt)
 					 errhint("The supported languages are listed in the pg_pltemplate system catalog.")));
 
 		/*
+		 * Check permission
+		 */
+		if (!superuser())
+			ereport(ERROR,
+					(errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
+					 errmsg("must be superuser to create custom procedural language")));
+
+		/*
 		 * Lookup the PL handler function and check that it is of the expected
 		 * return type
 		 */
@@ -227,7 +246,8 @@ CreateProceduralLanguage(CreatePLangStmt *stmt)
 			valOid = InvalidOid;
 
 		/* ok, create it */
-		create_proc_lang(languageName, handlerOid, valOid, stmt->pltrusted);
+		create_proc_lang(languageName, GetUserId(), handlerOid, valOid,
+						 stmt->pltrusted);
 	}
 }
 
@@ -236,7 +256,7 @@ CreateProceduralLanguage(CreatePLangStmt *stmt)
  */
 static void
 create_proc_lang(const char *languageName,
-				 Oid handlerOid, Oid valOid, bool trusted)
+				 Oid languageOwner, Oid handlerOid, Oid valOid, bool trusted)
 {
 	Relation	rel;
 	TupleDesc	tupDesc;
@@ -258,6 +278,7 @@ create_proc_lang(const char *languageName,
 
 	namestrcpy(&langname, languageName);
 	values[Anum_pg_language_lanname - 1] = NameGetDatum(&langname);
+	values[Anum_pg_language_lanowner - 1] = ObjectIdGetDatum(languageOwner);
 	values[Anum_pg_language_lanispl - 1] = BoolGetDatum(true);
 	values[Anum_pg_language_lanpltrusted - 1] = BoolGetDatum(trusted);
 	values[Anum_pg_language_lanplcallfoid - 1] = ObjectIdGetDatum(handlerOid);
@@ -277,6 +298,12 @@ create_proc_lang(const char *languageName,
 	myself.objectId = HeapTupleGetOid(tup);
 	myself.objectSubId = 0;
 
+	/* dependency on owner of language */
+	referenced.classId = AuthIdRelationId;
+	referenced.objectId = languageOwner;
+	referenced.objectSubId = 0;
+	recordSharedDependencyOn(&myself, &referenced, SHARED_DEPENDENCY_OWNER);
+
 	/* dependency on the PL handler function */
 	referenced.classId = ProcedureRelationId;
 	referenced.objectId = handlerOid;
@@ -325,6 +352,7 @@ find_language_template(const char *languageName)
 
 		result = (PLTemplate *) palloc0(sizeof(PLTemplate));
 		result->tmpltrusted = tmpl->tmpltrusted;
+		result->tmpldbacreate = tmpl->tmpldbacreate;
 
 		/* Remaining fields are variable-width so we need heap_getattr */
 		datum = heap_getattr(tup, Anum_pg_pltemplate_tmplhandler,
@@ -382,14 +410,6 @@ DropProceduralLanguage(DropPLangStmt *stmt)
 	ObjectAddress object;
 
 	/*
-	 * Check permission
-	 */
-	if (!superuser())
-		ereport(ERROR,
-				(errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
-				 errmsg("must be superuser to drop procedural language")));
-
-	/*
 	 * Translate the language name, check that the language exists
 	 */
 	languageName = case_translate_language_name(stmt->plname);
@@ -411,6 +431,13 @@ DropProceduralLanguage(DropPLangStmt *stmt)
 		return;
 	}
 
+	/*
+	 * Check permission
+	 */
+	if (!pg_language_ownercheck(HeapTupleGetOid(langTup), GetUserId()))
+		aclcheck_error(ACLCHECK_NOT_OWNER, ACL_KIND_LANGUAGE,
+					   languageName);
+
 	object.classId = LanguageRelationId;
 	object.objectId = HeapTupleGetOid(langTup);
 	object.objectSubId = 0;
@@ -478,11 +505,10 @@ RenameLanguage(const char *oldname, const char *newname)
 				(errcode(ERRCODE_DUPLICATE_OBJECT),
 				 errmsg("language \"%s\" already exists", newname)));
 
-	/* must be superuser, since we do not have owners for PLs */
-	if (!superuser())
-		ereport(ERROR,
-				(errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
-				 errmsg("must be superuser to rename procedural language")));
+	/* must be owner of PL */
+	if (!pg_language_ownercheck(HeapTupleGetOid(tup), GetUserId()))
+		aclcheck_error(ACLCHECK_NOT_OWNER, ACL_KIND_LANGUAGE,
+					   oldname);
 
 	/* rename */
 	namestrcpy(&(((Form_pg_language) GETSTRUCT(tup))->lanname), newname);
@@ -492,3 +518,87 @@ RenameLanguage(const char *oldname, const char *newname)
 	heap_close(rel, NoLock);
 	heap_freetuple(tup);
 }
+
+/*
+ * Change language owner
+ */
+void
+AlterLanguageOwner(const char *name, Oid newOwnerId)
+{
+	HeapTuple	tup;
+	Relation	rel;
+	Form_pg_language lanForm;
+
+	/* Translate name for consistency with CREATE */
+	name = case_translate_language_name(name);
+
+	rel = heap_open(LanguageRelationId, RowExclusiveLock);
+
+	tup = SearchSysCache(LANGNAME,
+						 CStringGetDatum(name),
+						 0, 0, 0);
+	if (!HeapTupleIsValid(tup))
+		ereport(ERROR,
+				(errcode(ERRCODE_UNDEFINED_OBJECT),
+				 errmsg("language \"%s\" does not exist", name)));
+	lanForm = (Form_pg_language) GETSTRUCT(tup);
+
+	/*
+	 * If the new owner is the same as the existing owner, consider the
+	 * command to have succeeded.  This is for dump restoration purposes.
+	 */
+	if (lanForm->lanowner != newOwnerId)
+	{
+		Datum		repl_val[Natts_pg_language];
+		char		repl_null[Natts_pg_language];
+		char		repl_repl[Natts_pg_language];
+		Acl		   *newAcl;
+		Datum		aclDatum;
+		bool		isNull;
+		HeapTuple	newtuple;
+
+		/* Otherwise, must be owner of the existing object */
+		if (!pg_language_ownercheck(HeapTupleGetOid(tup), GetUserId()))
+			aclcheck_error(ACLCHECK_NOT_OWNER, ACL_KIND_LANGUAGE,
+						   NameStr(lanForm->lanname));
+
+		/* Must be able to become new owner */
+		check_is_member_of_role(GetUserId(), newOwnerId);
+
+		memset(repl_null, ' ', sizeof(repl_null));
+		memset(repl_repl, ' ', sizeof(repl_repl));
+
+		repl_repl[Anum_pg_language_lanowner - 1] = 'r';
+		repl_val[Anum_pg_language_lanowner - 1] = ObjectIdGetDatum(newOwnerId);
+
+		/*
+		 * Determine the modified ACL for the new owner.  This is only
+		 * necessary when the ACL is non-null.
+		 */
+		aclDatum = SysCacheGetAttr(LANGNAME, tup,
+								   Anum_pg_language_lanacl,
+								   &isNull);
+		if (!isNull)
+		{
+			newAcl = aclnewowner(DatumGetAclP(aclDatum),
+								 lanForm->lanowner, newOwnerId);
+			repl_repl[Anum_pg_language_lanacl - 1] = 'r';
+			repl_val[Anum_pg_language_lanacl - 1] = PointerGetDatum(newAcl);
+		}
+
+		newtuple = heap_modifytuple(tup, RelationGetDescr(rel),
+									repl_val, repl_null, repl_repl);
+
+		simple_heap_update(rel, &newtuple->t_self, newtuple);
+		CatalogUpdateIndexes(rel, newtuple);
+
+		heap_freetuple(newtuple);
+
+		/* Update owner dependency reference */
+		changeDependencyOnOwner(LanguageRelationId, HeapTupleGetOid(tup),
+								newOwnerId);
+	}
+
+	ReleaseSysCache(tup);
+	heap_close(rel, RowExclusiveLock);
+}
diff --git a/src/backend/parser/gram.y b/src/backend/parser/gram.y
index 75404716d5a..8cad738a357 100644
--- a/src/backend/parser/gram.y
+++ b/src/backend/parser/gram.y
@@ -11,7 +11,7 @@
  *
  *
  * IDENTIFICATION
- *	  $PostgreSQL: pgsql/src/backend/parser/gram.y,v 2.583 2007/03/19 23:38:29 wieck Exp $
+ *	  $PostgreSQL: pgsql/src/backend/parser/gram.y,v 2.584 2007/03/26 16:58:39 tgl Exp $
  *
  * HISTORY
  *	  AUTHOR			DATE			MAJOR EVENT
@@ -4447,12 +4447,12 @@ RenameStmt: ALTER AGGREGATE func_name aggr_args RENAME TO name
 					n->newname = $6;
 					$$ = (Node *)n;
 				}
-			| ALTER LANGUAGE name RENAME TO name
+			| ALTER opt_procedural LANGUAGE name RENAME TO name
 				{
 					RenameStmt *n = makeNode(RenameStmt);
 					n->renameType = OBJECT_LANGUAGE;
-					n->subname = $3;
-					n->newname = $6;
+					n->subname = $4;
+					n->newname = $7;
 					$$ = (Node *)n;
 				}
 			| ALTER OPERATOR CLASS any_name USING access_method RENAME TO name
@@ -4654,6 +4654,14 @@ AlterOwnerStmt: ALTER AGGREGATE func_name aggr_args OWNER TO RoleId
 					n->newowner = $7;
 					$$ = (Node *)n;
 				}
+			| ALTER opt_procedural LANGUAGE name OWNER TO RoleId
+				{
+					AlterOwnerStmt *n = makeNode(AlterOwnerStmt);
+					n->objectType = OBJECT_LANGUAGE;
+					n->object = list_make1($4);
+					n->newowner = $7;
+					$$ = (Node *)n;
+				}
 			| ALTER OPERATOR any_operator '(' oper_argtypes ')' OWNER TO RoleId
 				{
 					AlterOwnerStmt *n = makeNode(AlterOwnerStmt);
diff --git a/src/backend/tcop/utility.c b/src/backend/tcop/utility.c
index be274a72f10..daf23fd0fe5 100644
--- a/src/backend/tcop/utility.c
+++ b/src/backend/tcop/utility.c
@@ -10,7 +10,7 @@
  *
  *
  * IDENTIFICATION
- *	  $PostgreSQL: pgsql/src/backend/tcop/utility.c,v 1.274 2007/03/13 00:33:42 tgl Exp $
+ *	  $PostgreSQL: pgsql/src/backend/tcop/utility.c,v 1.275 2007/03/26 16:58:39 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -1551,6 +1551,9 @@ CreateCommandTag(Node *parsetree)
 				case OBJECT_FUNCTION:
 					tag = "ALTER FUNCTION";
 					break;
+				case OBJECT_LANGUAGE:
+					tag = "ALTER LANGUAGE";
+					break;
 				case OBJECT_OPERATOR:
 					tag = "ALTER OPERATOR";
 					break;