postgresql_owner: add trust_input parameter (#198)

* postgresql_owner: add trust_input parameter, allow to pass values containing dots to some parameters

* add changelog fragment

* fix CI

* fix CI

changelogs/fragments/198-postgresql_owner_add_trust_input_parameter.yml

@@ -0,0 +1,2 @@
+minor_changes:
+- postgresql_owner - add the ``trust_input`` parameter (https://github.com/ansible-collections/community.general/pull/198).

plugins/modules/database/postgresql/postgresql_owner.py

@@ -70,6 +70,11 @@
     - Permissions checking for SQL commands is carried out as though
       the session_role were the one that had logged in originally.
     type: str
+  trust_input:
+    description:
+    - If C(no), check whether values of some parameters are potentially dangerous.
+    type: bool
+    default: yes
 seealso:
 - module: postgresql_user
 - module: postgresql_privs
@@ -147,7 +152,10 @@
     pass
 
 from ansible.module_utils.basic import AnsibleModule
-from ansible_collections.community.general.plugins.module_utils.database import pg_quote_identifier
+from ansible_collections.community.general.plugins.module_utils.database import (
+    check_input,
+    pg_quote_identifier,
+)
 from ansible_collections.community.general.plugins.module_utils.postgres import (
     connect_to_db,
     exec_sql,
@@ -218,7 +226,7 @@ def reassign(self, old_owners, fail_on_role):
         roles = []
         for r in old_owners:
             if self.check_role_exists(r, fail_on_role):
-                roles.append(pg_quote_identifier(r, 'role'))
+                roles.append('"%s"' % r)
 
         # Roles do not exist, nothing to do, exit:
         if not roles:
@@ -228,7 +236,7 @@ def reassign(self, old_owners, fail_on_role):
 
         query = ['REASSIGN OWNED BY']
         query.append(old_owners)
-        query.append('TO %s' % pg_quote_identifier(self.role, 'role'))
+        query.append('TO "%s"' % self.role)
         query = ' '.join(query)
 
         self.changed = exec_sql(self, query, return_bool=True)
@@ -323,50 +331,47 @@ def __is_owner(self):
 
     def __set_db_owner(self):
         """Set the database owner."""
-        query = "ALTER DATABASE %s OWNER TO %s" % (pg_quote_identifier(self.obj_name, 'database'),
-                                                   pg_quote_identifier(self.role, 'role'))
+        query = 'ALTER DATABASE "%s" OWNER TO "%s"' % (self.obj_name, self.role)
         self.changed = exec_sql(self, query, return_bool=True)
 
     def __set_func_owner(self):
         """Set the function owner."""
-        query = "ALTER FUNCTION %s OWNER TO %s" % (self.obj_name,
-                                                   pg_quote_identifier(self.role, 'role'))
+        query = 'ALTER FUNCTION %s OWNER TO "%s"' % (self.obj_name, self.role)
         self.changed = exec_sql(self, query, return_bool=True)
 
     def __set_seq_owner(self):
         """Set the sequence owner."""
-        query = "ALTER SEQUENCE %s OWNER TO %s" % (pg_quote_identifier(self.obj_name, 'table'),
-                                                   pg_quote_identifier(self.role, 'role'))
+        query = 'ALTER SEQUENCE %s OWNER TO "%s"' % (pg_quote_identifier(self.obj_name, 'table'),
+                                                     self.role)
         self.changed = exec_sql(self, query, return_bool=True)
 
     def __set_schema_owner(self):
         """Set the schema owner."""
-        query = "ALTER SCHEMA %s OWNER TO %s" % (pg_quote_identifier(self.obj_name, 'schema'),
-                                                 pg_quote_identifier(self.role, 'role'))
+        query = 'ALTER SCHEMA %s OWNER TO "%s"' % (pg_quote_identifier(self.obj_name, 'schema'),
+                                                   self.role)
         self.changed = exec_sql(self, query, return_bool=True)
 
     def __set_table_owner(self):
         """Set the table owner."""
-        query = "ALTER TABLE %s OWNER TO %s" % (pg_quote_identifier(self.obj_name, 'table'),
-                                                pg_quote_identifier(self.role, 'role'))
+        query = 'ALTER TABLE %s OWNER TO "%s"' % (pg_quote_identifier(self.obj_name, 'table'),
+                                                  self.role)
         self.changed = exec_sql(self, query, return_bool=True)
 
     def __set_tablespace_owner(self):
         """Set the tablespace owner."""
-        query = "ALTER TABLESPACE %s OWNER TO %s" % (pg_quote_identifier(self.obj_name, 'database'),
-                                                     pg_quote_identifier(self.role, 'role'))
+        query = 'ALTER TABLESPACE "%s" OWNER TO "%s"' % (self.obj_name, self.role)
         self.changed = exec_sql(self, query, return_bool=True)
 
     def __set_view_owner(self):
         """Set the view owner."""
-        query = "ALTER VIEW %s OWNER TO %s" % (pg_quote_identifier(self.obj_name, 'table'),
-                                               pg_quote_identifier(self.role, 'role'))
+        query = 'ALTER VIEW %s OWNER TO "%s"' % (pg_quote_identifier(self.obj_name, 'table'),
+                                                 self.role)
         self.changed = exec_sql(self, query, return_bool=True)
 
     def __set_mat_view_owner(self):
         """Set the materialized view owner."""
-        query = "ALTER MATERIALIZED VIEW %s OWNER TO %s" % (pg_quote_identifier(self.obj_name, 'table'),
-                                                            pg_quote_identifier(self.role, 'role'))
+        query = 'ALTER MATERIALIZED VIEW %s OWNER TO "%s"' % (pg_quote_identifier(self.obj_name, 'table'),
+                                                              self.role)
         self.changed = exec_sql(self, query, return_bool=True)
 
     def __role_exists(self, role):
@@ -392,6 +397,7 @@ def main():
         fail_on_role=dict(type='bool', default=True),
         db=dict(type='str', aliases=['login_db']),
         session_role=dict(type='str'),
+        trust_input=dict(type='bool', default=True),
     )
     module = AnsibleModule(
         argument_spec=argument_spec,
@@ -409,6 +415,11 @@ def main():
     obj_type = module.params['obj_type']
     reassign_owned_by = module.params['reassign_owned_by']
     fail_on_role = module.params['fail_on_role']
+    session_role = module.params['session_role']
+    trust_input = module.params['trust_input']
+    if not trust_input:
+        # Check input for potentially dangerous elements:
+        check_input(module, new_owner, obj_name, reassign_owned_by, session_role)
 
     conn_params = get_conn_params(module, module.params)
     db_connection = connect_to_db(module, conn_params, autocommit=False)

tests/integration/targets/postgresql_owner/defaults/main.yml

@@ -1 +1,3 @@
 test_tablespace_path: "/ssd"
+
+dangerous_name: 'curious.anonymous"; SELECT * FROM information_schema.tables; --'