fix permissions

Author: zilder

Makefile

@@ -12,11 +12,13 @@ EXTVERSION = 1.0
 DATA_built = $(EXTENSION)--$(EXTVERSION).sql
 PGFILEDESC = "pg_pathman - partitioning tool"
 
-REGRESS = pathman_basic \
-		  pathman_runtime_nodes \
-		  pathman_callbacks \
-		  pathman_domains \
-		  pathman_foreign_keys
+# REGRESS = pathman_basic \
+# 		  pathman_runtime_nodes \
+# 		  pathman_callbacks \
+# 		  pathman_domains \
+# 		  pathman_foreign_keys \
+# 		  pathman_permissions
+REGRESS = pathman_permissions
 EXTRA_REGRESS_OPTS=--temp-config=$(top_srcdir)/$(subdir)/conf.add
 EXTRA_CLEAN = $(EXTENSION)--$(EXTVERSION).sql ./isolation_output
 

expected/pathman_permissions.out

@@ -0,0 +1,61 @@
+\set VERBOSITY terse
+CREATE EXTENSION pg_pathman;
+CREATE ROLE user1 LOGIN;
+CREATE ROLE user2 LOGIN;
+SET ROLE user1;
+CREATE TABLE user1_table(id serial, a int);
+INSERT INTO user1_table SELECT g, g FROM generate_series(1, 20) as g;
+ALTER DEFAULT PRIVILEGES IN SCHEMA public GRANT SELECT, INSERT, UPDATE, DELETE ON TABLES TO public;
+SET ROLE user2;
+/* Not owner and not superuser cannot create partitions */
+SELECT create_range_partitions('user1_table', 'id', 1, 10, 2);
+ERROR:  permission denied for relation user1_table
+SET ROLE user1;
+/* Owner can */
+SELECT create_range_partitions('user1_table', 'id', 1, 10, 2);
+NOTICE:  sequence "user1_table_seq" does not exist, skipping
+ create_range_partitions 
+-------------------------
+                       2
+(1 row)
+
+SET ROLE user2;
+/* Try to change partitioning parameters for user1_table */
+SELECT set_enable_parent('user1_table', true);
+ERROR:  Only table owner or superuser can change partitioning configuration
+SELECT set_auto('user1_table', false);
+ERROR:  Only table owner or superuser can change partitioning configuration
+/*
+ * Check that user is able to propagate partitions by inserting rows that
+ * doesn't fit into range
+ */
+INSERT INTO user1_table (id, a) VALUES (25, 0);
+ERROR:  permission denied for relation user1_table
+SET ROLE user1;
+SELECT drop_partitions('test1_table');
+ERROR:  relation "test1_table" does not exist at character 24
+SET ROLE user2;
+/* Test ddl event trigger */
+CREATE TABLE user2_table(id serial);
+SELECT create_hash_partitions('user2_table', 'id', 3);
+ create_hash_partitions 
+------------------------
+                      3
+(1 row)
+
+INSERT INTO user2_table SELECT generate_series(1, 30);
+SELECT drop_partitions('user2_table');
+NOTICE:  function public.user2_table_upd_trig_func() does not exist, skipping
+NOTICE:  9 rows copied from user2_table_0
+NOTICE:  11 rows copied from user2_table_1
+NOTICE:  10 rows copied from user2_table_2
+ drop_partitions 
+-----------------
+               3
+(1 row)
+
+RESET ROLE;
+DROP OWNED BY user1;
+DROP OWNED BY user2;
+DROP USER user1;
+DROP USER user2;

hash.sql

@@ -27,8 +27,6 @@ DECLARE
 	v_init_callback		REGPROCEDURE;
 
 BEGIN
-	PERFORM @extschema@.check_permissions(parent_relid);
-
 	IF partition_data = true THEN
 		/* Acquire data modification lock */
 		PERFORM @extschema@.prevent_relation_modification(parent_relid);

init.sql

@@ -49,24 +49,24 @@ TO public;
 /*
  * Check if current user can alter/drop specified relation
  */
-CREATE OR REPLACE FUNCTION @extschema@.can_manage_relation(relation regclass)
-RETURNS BOOL AS 'pg_pathman', 'can_manage_relation' LANGUAGE C STRICT;
+CREATE OR REPLACE FUNCTION @extschema@.check_security_policy(relation regclass)
+RETURNS BOOL AS 'pg_pathman', 'check_security_policy' LANGUAGE C STRICT;
 
 /*
  * Check user permissions. If permission denied then throw an error.
  */
-CREATE OR REPLACE FUNCTION @extschema@.check_permissions(relation regclass)
-RETURNS BOOL AS 'pg_pathman', 'check_permissions' LANGUAGE C STRICT;
+-- CREATE OR REPLACE FUNCTION @extschema@.check_permissions(relation regclass)
+-- RETURNS BOOL AS 'pg_pathman', 'check_permissions' LANGUAGE C STRICT;
 
 /*
  * Row security policy to restrict partitioning operations to owner and
  * superusers only
  */
 CREATE POLICY deny_modification ON @extschema@.pathman_config
-FOR ALL USING (can_manage_relation(partrel));
+FOR ALL USING (check_security_policy(partrel));
 
 CREATE POLICY deny_modification ON @extschema@.pathman_config_params
-FOR ALL USING (can_manage_relation(partrel));
+FOR ALL USING (check_security_policy(partrel));
 
 CREATE POLICY allow_select ON @extschema@.pathman_config FOR SELECT USING (true);
 
@@ -129,7 +129,7 @@ CREATE OR REPLACE FUNCTION @extschema@.pathman_set_param(
 RETURNS VOID AS
 $$
 BEGIN
-	PERFORM @extschema@.check_permissions(relation);
+	-- PERFORM @extschema@.check_permissions(relation);
 
 	EXECUTE format('INSERT INTO @extschema@.pathman_config_params
 					(partrel, %1$s) VALUES ($1, $2)
@@ -336,7 +336,7 @@ CREATE OR REPLACE FUNCTION @extschema@.disable_pathman_for(
 RETURNS VOID AS
 $$
 BEGIN
-	PERFORM @extschema@.check_permissions(parent_relid);
+	-- PERFORM @extschema@.check_permissions(parent_relid);
 
 	DELETE FROM @extschema@.pathman_config WHERE partrel = parent_relid;
 	PERFORM @extschema@.drop_triggers(parent_relid);
@@ -468,34 +468,29 @@ $$
 LANGUAGE plpgsql;
 
 /*
- * DDL trigger that deletes entry from pathman_config table.
+ * DDL trigger that removes entry from pathman_config table.
  */
 CREATE OR REPLACE FUNCTION @extschema@.pathman_ddl_trigger_func()
 RETURNS event_trigger AS
 $$
 DECLARE
 	obj				record;
 	pg_class_oid	oid;
+	relids			regclass[];
 BEGIN
 	pg_class_oid = 'pg_catalog.pg_class'::regclass;
 
-	/* Handle 'DROP TABLE' events */
-	WITH to_be_deleted AS (
-		SELECT cfg.partrel AS rel FROM pg_event_trigger_dropped_objects() AS events
-		JOIN @extschema@.pathman_config AS cfg ON cfg.partrel::oid = events.objid
-		WHERE events.classid = pg_class_oid
-	)
-	DELETE FROM @extschema@.pathman_config
-	WHERE partrel IN (SELECT rel FROM to_be_deleted);
+	/* Find relids to remove from config */
+	SELECT array_agg(cfg.partrel) INTO relids
+	FROM pg_event_trigger_dropped_objects() AS events
+	JOIN @extschema@.pathman_config AS cfg ON cfg.partrel::oid = events.objid
+	WHERE events.classid = pg_class_oid;
+
+	/* Cleanup pathman_config */
+	DELETE FROM @extschema@.pathman_config WHERE partrel = ANY(relids);
 
 	/* Cleanup params table too */
-	WITH to_be_deleted AS (
-		SELECT cfg.partrel AS rel FROM pg_event_trigger_dropped_objects() AS events
-		JOIN @extschema@.pathman_config_params AS cfg ON cfg.partrel::oid = events.objid
-		WHERE events.classid = pg_class_oid
-	)
-	DELETE FROM @extschema@.pathman_config_params
-	WHERE partrel IN (SELECT rel FROM to_be_deleted);
+	DELETE FROM @extschema@.pathman_config_params WHERE partrel = ANY(relids);
 END
 $$
 LANGUAGE plpgsql;
@@ -530,7 +525,7 @@ DECLARE
 	v_relkind		CHAR;
 
 BEGIN
-	PERFORM @extschema@.check_permissions(parent_relid);
+	-- PERFORM @extschema@.check_permissions(parent_relid);
 
 	/* Drop trigger first */
 	PERFORM @extschema@.drop_triggers(parent_relid);

range.sql

@@ -95,8 +95,6 @@ DECLARE
 	i					INTEGER;
 
 BEGIN
-	PERFORM @extschema@.check_permissions(parent_relid);
-
 	IF partition_data = true THEN
 		/* Acquire data modification lock */
 		PERFORM @extschema@.prevent_relation_modification(parent_relid);
@@ -208,8 +206,6 @@ DECLARE
 	i					INTEGER;
 
 BEGIN
-	PERFORM @extschema@.check_permissions(parent_relid);
-
 	IF partition_data = true THEN
 		/* Acquire data modification lock */
 		PERFORM @extschema@.prevent_relation_modification(parent_relid);
@@ -316,8 +312,6 @@ DECLARE
 	part_count		INTEGER := 0;
 
 BEGIN
-	PERFORM @extschema@.check_permissions(parent_relid);
-
 	IF partition_data = true THEN
 		/* Acquire data modification lock */
 		PERFORM @extschema@.prevent_relation_modification(parent_relid);
@@ -390,8 +384,6 @@ DECLARE
 	part_count		INTEGER := 0;
 
 BEGIN
-	PERFORM @extschema@.check_permissions(parent_relid);
-
 	IF partition_data = true THEN
 		/* Acquire data modification lock */
 		PERFORM @extschema@.prevent_relation_modification(parent_relid);

sql/pathman_permissions.sql

@@ -1,17 +1,53 @@
-\SET VERBOSITY terse
+\set VERBOSITY terse
 
 CREATE EXTENSION pg_pathman;
 CREATE ROLE user1 LOGIN;
 CREATE ROLE user2 LOGIN;
 
-\SET ROLE user1
+SET ROLE user1;
 
 CREATE TABLE user1_table(id serial, a int);
-SELECT create_hash_partitioning('user1_table', 'id', 3);
+INSERT INTO user1_table SELECT g, g FROM generate_series(1, 20) as g;
 
-ALTER DEFAULT PRIVILEGES IN SCHEMA public GRANT INSERT ON TABLES TO public;
+ALTER DEFAULT PRIVILEGES IN SCHEMA public GRANT SELECT, INSERT, UPDATE, DELETE ON TABLES TO public;
 
-\SET ROLE user2
+SET ROLE user2;
 
-/* Trying to change partitioning parameters for user1_table */
+/* Not owner and not superuser cannot create partitions */
+SELECT create_range_partitions('user1_table', 'id', 1, 10, 2);
+
+SET ROLE user1;
+
+/* Owner can */
+SELECT create_range_partitions('user1_table', 'id', 1, 10, 2);
+
+SET ROLE user2;
+
+/* Try to change partitioning parameters for user1_table */
 SELECT set_enable_parent('user1_table', true);
+SELECT set_auto('user1_table', false);
+
+/*
+ * Check that user is able to propagate partitions by inserting rows that
+ * doesn't fit into range
+ */
+INSERT INTO user1_table (id, a) VALUES (25, 0);
+
+SET ROLE user1;
+
+SELECT drop_partitions('test1_table');
+
+SET ROLE user2;
+
+/* Test ddl event trigger */
+CREATE TABLE user2_table(id serial);
+SELECT create_hash_partitions('user2_table', 'id', 3);
+INSERT INTO user2_table SELECT generate_series(1, 30);
+SELECT drop_partitions('user2_table');
+
+RESET ROLE;
+
+DROP OWNED BY user1;
+DROP OWNED BY user2;
+DROP USER user1;
+DROP USER user2;

src/pl_funcs.c

@@ -28,7 +28,6 @@
 #include "utils/lsyscache.h"
 #include "utils/syscache.h"
 
-static bool can_manage_relation_internal(Oid relid);
 
 /* Function declarations */
 
@@ -60,8 +59,7 @@ PG_FUNCTION_INFO_V1( prevent_relation_modification );
 PG_FUNCTION_INFO_V1( validate_on_part_init_callback_pl );
 PG_FUNCTION_INFO_V1( invoke_on_partition_created_callback );
 
-PG_FUNCTION_INFO_V1( can_manage_relation );
-PG_FUNCTION_INFO_V1( check_permissions );
+PG_FUNCTION_INFO_V1( check_security_policy );
 
 PG_FUNCTION_INFO_V1( debug_capture );
 
@@ -858,45 +856,11 @@ invoke_on_partition_created_callback(PG_FUNCTION_ARGS)
  * check user permissions in order to let other users.
  */
 Datum
-can_manage_relation(PG_FUNCTION_ARGS)
+check_security_policy(PG_FUNCTION_ARGS)
 {
 	Oid 		relid = PG_GETARG_OID(0);
 
-	PG_RETURN_BOOL(can_manage_relation_internal(relid));
-}
-
-static bool
-can_manage_relation_internal(Oid relid)
-{
-	Oid 		owner;
-
-	/*
-	 * If user has superuser privileges then he or she can do whatever wants
-	 */
-	if (superuser())
-		return true;
-
-	/* Otherwise check if user is owner of the relation */
-	owner = get_rel_owner(relid);
-	if (owner == GetUserId())
-		return true;
-
-	return false;
-}
-
-/*
- * Check user permissions. If permission denied then throw an error.
- */
-Datum
-check_permissions(PG_FUNCTION_ARGS)
-{
-	Oid 		relid = PG_GETARG_OID(0);
-
-	if (!can_manage_relation_internal(relid))
-		elog(ERROR,
-			 "Only table owner or superuser can change partitioning configuration");
-
-	PG_RETURN_VOID();
+	PG_RETURN_BOOL(check_security_policy_internal(relid));
 }
 
 

src/utils.c

@@ -743,3 +743,37 @@ validate_on_part_init_cb(Oid procid, bool emit_error)
 
 	return is_ok;
 }
+
+/*
+ * Check if user can alter/drop specified relation. This function is used to
+ * make sure that current user can change pg_pathman's config. Returns true
+ * if user can manage relation, false otherwise.
+ *
+ * XXX currently we just check if user is a table owner. Probably it's better to
+ * check user permissions in order to let other users.
+ */
+bool
+check_security_policy_internal(Oid relid)
+{
+	Oid 		owner;
+
+	/*
+	 * If user has superuser privileges then he or she can do whatever wants
+	 */
+	if (superuser())
+		return true;
+
+	/*
+	 * Sometimes the relation doesn't exist anymore but there is still a record
+	 * in config. It for example happens in event trigger function. So we
+	 * should be able to remove this record
+	 */
+	if ((owner = get_rel_owner(relid)) == InvalidOid)
+		return true;
+
+	/* Check if current user is an owner of the relation */
+	if (owner != GetUserId())
+		elog(ERROR, "Only table owner or superuser can change partitioning configuration");
+
+	return true;
+}