Add large object access control.

A new system catalog pg_largeobject_metadata manages
ownership and access privileges of large objects.

KaiGai Kohei, reviewed by Jaime Casanova.

5 files changed, 103 insertions, 34 deletions

diff --git a/src/bin/initdb/initdb.c b/src/bin/initdb/initdb.c
index de0aba36a86..6f1f211c141 100644
--- a/src/bin/initdb/initdb.c
+++ b/src/bin/initdb/initdb.c
@@ -42,7 +42,7 @@
  * Portions Copyright (c) 1994, Regents of the University of California
  * Portions taken from FreeBSD.
  *
- * $PostgreSQL: pgsql/src/bin/initdb/initdb.c,v 1.177 2009/11/14 15:39:36 mha Exp $
+ * $PostgreSQL: pgsql/src/bin/initdb/initdb.c,v 1.178 2009/12/11 03:34:56 itagaki Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -1783,6 +1783,7 @@ setup_privileges(void)
 		"  WHERE relkind IN ('r', 'v', 'S') AND relacl IS NULL;\n",
 		"GRANT USAGE ON SCHEMA pg_catalog TO PUBLIC;\n",
 		"GRANT CREATE, USAGE ON SCHEMA public TO PUBLIC;\n",
+		"REVOKE ALL ON pg_largeobject FROM PUBLIC;\n",
 		NULL
 	};
 
diff --git a/src/bin/pg_dump/dumputils.c b/src/bin/pg_dump/dumputils.c
index c7cfa2a6212..45559665aee 100644
--- a/src/bin/pg_dump/dumputils.c
+++ b/src/bin/pg_dump/dumputils.c
@@ -8,7 +8,7 @@
  * Portions Copyright (c) 1996-2009, PostgreSQL Global Development Group
  * Portions Copyright (c) 1994, Regents of the University of California
  *
- * $PostgreSQL: pgsql/src/bin/pg_dump/dumputils.c,v 1.51 2009/10/12 23:41:43 tgl Exp $
+ * $PostgreSQL: pgsql/src/bin/pg_dump/dumputils.c,v 1.52 2009/12/11 03:34:56 itagaki Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -862,6 +862,11 @@ do { \
 		CONVERT_PRIV('U', "USAGE");
 	else if (strcmp(type, "SERVER") == 0)
 		CONVERT_PRIV('U', "USAGE");
+	else if (strcmp(type, "LARGE OBJECT") == 0)
+	{
+		CONVERT_PRIV('r', "SELECT");
+		CONVERT_PRIV('w', "UPDATE");
+	}
 	else
 		abort();
 
diff --git a/src/bin/pg_dump/pg_dump.c b/src/bin/pg_dump/pg_dump.c
index a753976139c..7af461bd5a6 100644
--- a/src/bin/pg_dump/pg_dump.c
+++ b/src/bin/pg_dump/pg_dump.c
@@ -12,7 +12,7 @@
  *	by PostgreSQL
  *
  * IDENTIFICATION
- *	  $PostgreSQL: pgsql/src/bin/pg_dump/pg_dump.c,v 1.554 2009/12/07 05:22:22 tgl Exp $
+ *	  $PostgreSQL: pgsql/src/bin/pg_dump/pg_dump.c,v 1.555 2009/12/11 03:34:56 itagaki Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -2045,7 +2045,9 @@ dumpBlobs(Archive *AH, void *arg)
 
 /*
  * dumpBlobComments
- *	dump all blob comments
+ *	dump all blob properties.
+ *  It has "BLOB COMMENTS" tag due to the historical reason, but note
+ *  that it is the routine to dump all the properties of blobs.
  *
  * Since we don't provide any way to be selective about dumping blobs,
  * there's no need to be selective about their comments either.  We put
@@ -2056,30 +2058,35 @@ dumpBlobComments(Archive *AH, void *arg)
 {
 	const char *blobQry;
 	const char *blobFetchQry;
-	PQExpBuffer commentcmd = createPQExpBuffer();
+	PQExpBuffer cmdQry = createPQExpBuffer();
 	PGresult   *res;
 	int			i;
 
 	if (g_verbose)
-		write_msg(NULL, "saving large object comments\n");
+		write_msg(NULL, "saving large object properties\n");
 
 	/* Make sure we are in proper schema */
 	selectSourceSchema("pg_catalog");
 
 	/* Cursor to get all BLOB comments */
-	if (AH->remoteVersion >= 70300)
+	if (AH->remoteVersion >= 80500)
+		blobQry = "DECLARE blobcmt CURSOR FOR SELECT oid, "
+			"obj_description(oid, 'pg_largeobject'), "
+			"pg_get_userbyid(lomowner), lomacl "
+			"FROM pg_largeobject_metadata";
+	else if (AH->remoteVersion >= 70300)
 		blobQry = "DECLARE blobcmt CURSOR FOR SELECT loid, "
-			"obj_description(loid, 'pg_largeobject') "
+			"obj_description(loid, 'pg_largeobject'), NULL, NULL "
 			"FROM (SELECT DISTINCT loid FROM "
 			"pg_description d JOIN pg_largeobject l ON (objoid = loid) "
 			"WHERE classoid = 'pg_largeobject'::regclass) ss";
 	else if (AH->remoteVersion >= 70200)
 		blobQry = "DECLARE blobcmt CURSOR FOR SELECT loid, "
-			"obj_description(loid, 'pg_largeobject') "
+			"obj_description(loid, 'pg_largeobject'), NULL, NULL "
 			"FROM (SELECT DISTINCT loid FROM pg_largeobject) ss";
 	else if (AH->remoteVersion >= 70100)
 		blobQry = "DECLARE blobcmt CURSOR FOR SELECT loid, "
-			"obj_description(loid) "
+			"obj_description(loid), NULL, NULL "
 			"FROM (SELECT DISTINCT loid FROM pg_largeobject) ss";
 	else
 		blobQry = "DECLARE blobcmt CURSOR FOR SELECT oid, "
@@ -2087,7 +2094,7 @@ dumpBlobComments(Archive *AH, void *arg)
 			"		SELECT description "
 			"		FROM pg_description pd "
 			"		WHERE pd.objoid=pc.oid "
-			"	) "
+			"	), NULL, NULL "
 			"FROM pg_class pc WHERE relkind = 'l'";
 
 	res = PQexec(g_conn, blobQry);
@@ -2107,22 +2114,51 @@ dumpBlobComments(Archive *AH, void *arg)
 		/* Process the tuples, if any */
 		for (i = 0; i < PQntuples(res); i++)
 		{
-			Oid			blobOid;
-			char	   *comment;
+			Oid			blobOid = atooid(PQgetvalue(res, i, 0));
+			char	   *lo_comment = PQgetvalue(res, i, 1);
+			char	   *lo_owner = PQgetvalue(res, i, 2);
+			char	   *lo_acl = PQgetvalue(res, i, 3);
+			char		lo_name[32];
 
-			/* ignore blobs without comments */
-			if (PQgetisnull(res, i, 1))
-				continue;
+			resetPQExpBuffer(cmdQry);
 
-			blobOid = atooid(PQgetvalue(res, i, 0));
-			comment = PQgetvalue(res, i, 1);
+			/* comment on the blob */
+			if (!PQgetisnull(res, i, 1))
+			{
+				appendPQExpBuffer(cmdQry,
+								  "COMMENT ON LARGE OBJECT %u IS ", blobOid);
+				appendStringLiteralAH(cmdQry, lo_comment, AH);
+				appendPQExpBuffer(cmdQry, ";\n");
+			}
+
+			/* dump blob ownership, if necessary */
+			if (!PQgetisnull(res, i, 2))
+			{
+				appendPQExpBuffer(cmdQry,
+								  "ALTER LARGE OBJECT %u OWNER TO %s;\n",
+								  blobOid, lo_owner);
+			}
 
-			printfPQExpBuffer(commentcmd, "COMMENT ON LARGE OBJECT %u IS ",
-							  blobOid);
-			appendStringLiteralAH(commentcmd, comment, AH);
-			appendPQExpBuffer(commentcmd, ";\n");
+			/* dump blob privileges, if necessary */
+			if (!PQgetisnull(res, i, 3) &&
+				!dataOnly && !aclsSkip)
+			{
+				snprintf(lo_name, sizeof(lo_name), "%u", blobOid);
+				if (!buildACLCommands(lo_name, NULL, "LARGE OBJECT",
+									  lo_acl, lo_owner, "",
+									  AH->remoteVersion, cmdQry))
+				{
+					write_msg(NULL, "could not parse ACL (%s) for "
+							  "large object %u", lo_acl, blobOid);
+					exit_nicely();
+				}
+			}
 
-			archputs(commentcmd->data, AH);
+			if (cmdQry->len > 0)
+			{
+				appendPQExpBuffer(cmdQry, "\n");
+				archputs(cmdQry->data, AH);
+			}
 		}
 	} while (PQntuples(res) > 0);
 
@@ -2130,7 +2166,7 @@ dumpBlobComments(Archive *AH, void *arg)
 
 	archputs("\n", AH);
 
-	destroyPQExpBuffer(commentcmd);
+	destroyPQExpBuffer(cmdQry);
 
 	return 1;
 }
diff --git a/src/bin/psql/large_obj.c b/src/bin/psql/large_obj.c
index e4da1d3e962..c77077f9870 100644
--- a/src/bin/psql/large_obj.c
+++ b/src/bin/psql/large_obj.c
@@ -3,7 +3,7 @@
  *
  * Copyright (c) 2000-2009, PostgreSQL Global Development Group
  *
- * $PostgreSQL: pgsql/src/bin/psql/large_obj.c,v 1.52 2009/01/01 17:23:55 momjian Exp $
+ * $PostgreSQL: pgsql/src/bin/psql/large_obj.c,v 1.53 2009/12/11 03:34:56 itagaki Exp $
  */
 #include "postgres_fe.h"
 #include "large_obj.h"
@@ -278,13 +278,28 @@ do_lo_list(void)
 	char		buf[1024];
 	printQueryOpt myopt = pset.popt;
 
-	snprintf(buf, sizeof(buf),
-			 "SELECT loid as \"%s\",\n"
-		   "  pg_catalog.obj_description(loid, 'pg_largeobject') as \"%s\"\n"
-			 "FROM (SELECT DISTINCT loid FROM pg_catalog.pg_largeobject) x\n"
-			 "ORDER BY 1",
-			 gettext_noop("ID"),
-			 gettext_noop("Description"));
+	if (pset.sversion >= 80500)
+	{
+		snprintf(buf, sizeof(buf),
+				 "SELECT oid as \"%s\",\n"
+				 "  pg_catalog.pg_get_userbyid(lomowner) as \"%s\",\n"
+				 "  pg_catalog.obj_description(oid, 'pg_largeobject') as \"%s\"\n"
+				 "  FROM pg_catalog.pg_largeobject_metadata "
+				 "  ORDER BY oid",
+				 gettext_noop("ID"),
+				 gettext_noop("Owner"),
+				 gettext_noop("Description"));
+	}
+	else
+	{
+		snprintf(buf, sizeof(buf),
+				 "SELECT loid as \"%s\",\n"
+				 "  pg_catalog.obj_description(loid, 'pg_largeobject') as \"%s\"\n"
+				 "FROM (SELECT DISTINCT loid FROM pg_catalog.pg_largeobject) x\n"
+				 "ORDER BY 1",
+				 gettext_noop("ID"),
+				 gettext_noop("Description"));
+	}
 
 	res = PSQLexec(buf, false);
 	if (!res)
diff --git a/src/bin/psql/tab-complete.c b/src/bin/psql/tab-complete.c
index 7cd07574151..10734fe00c4 100644
--- a/src/bin/psql/tab-complete.c
+++ b/src/bin/psql/tab-complete.c
@@ -3,7 +3,7 @@
  *
  * Copyright (c) 2000-2009, PostgreSQL Global Development Group
  *
- * $PostgreSQL: pgsql/src/bin/psql/tab-complete.c,v 1.187 2009/10/13 21:04:01 tgl Exp $
+ * $PostgreSQL: pgsql/src/bin/psql/tab-complete.c,v 1.188 2009/12/11 03:34:56 itagaki Exp $
  */
 
 /*----------------------------------------------------------------------
@@ -691,7 +691,7 @@ psql_completion(char *text, int start, int end)
 	{
 		static const char *const list_ALTER[] =
 		{"AGGREGATE", "CONVERSION", "DATABASE", "DOMAIN", "FOREIGN DATA WRAPPER", "FUNCTION",
-			"GROUP", "INDEX", "LANGUAGE", "OPERATOR", "ROLE", "SCHEMA", "SERVER", "SEQUENCE", "TABLE",
+			"GROUP", "INDEX", "LANGUAGE", "LARGE OBJECT", "OPERATOR", "ROLE", "SCHEMA", "SERVER", "SEQUENCE", "TABLE",
 		"TABLESPACE", "TEXT SEARCH", "TRIGGER", "TYPE", "USER", "USER MAPPING FOR", "VIEW", NULL};
 
 		COMPLETE_WITH_LIST(list_ALTER);
@@ -760,6 +760,17 @@ psql_completion(char *text, int start, int end)
 		COMPLETE_WITH_LIST(list_ALTERLANGUAGE);
 	}
 
+	/* ALTER LARGE OBJECT <oid> */
+	else if (pg_strcasecmp(prev4_wd, "ALTER") == 0 &&
+			 pg_strcasecmp(prev3_wd, "LARGE") == 0 &&
+			 pg_strcasecmp(prev2_wd, "OBJECT") == 0)
+	{
+		static const char *const list_ALTERLARGEOBJECT[] =
+		{"OWNER TO", NULL};
+
+		COMPLETE_WITH_LIST(list_ALTERLARGEOBJECT);
+	}
+
 	/* ALTER USER,ROLE <name> */
 	else if (pg_strcasecmp(prev3_wd, "ALTER") == 0 &&
 			 !(pg_strcasecmp(prev2_wd, "USER") == 0 && pg_strcasecmp(prev_wd, "MAPPING") == 0) &&
@@ -1732,6 +1743,7 @@ psql_completion(char *text, int start, int end)
 								   " UNION SELECT 'FOREIGN SERVER'"
 								   " UNION SELECT 'FUNCTION'"
 								   " UNION SELECT 'LANGUAGE'"
+								   " UNION SELECT 'LARGE OBJECT'"
 								   " UNION SELECT 'SCHEMA'"
 								   " UNION SELECT 'TABLESPACE'");