postgresql_schema: add trust_input parameter (#259)

* postgresql_schema: add trust_input parameter

* add changelog fragment

changelogs/fragments/259-postgresql_schema_add_trust_input_parameter.yml

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

plugins/modules/database/postgresql/postgresql_schema.py

@@ -69,6 +69,11 @@ options:
       - If the file exists, the server's certificate will be verified to be signed by one of these authorities.
     type: str
     aliases: [ ssl_rootcert ]
+  trust_input:
+    description:
+    - If C(no), check whether values of some parameters are potentially dangerous.
+    type: bool
+    default: yes
 seealso:
 - name: PostgreSQL schemas
   description: General information about PostgreSQL schemas.
@@ -136,7 +141,11 @@ from ansible_collections.community.general.plugins.module_utils.postgres import
     get_conn_params,
     postgres_common_argument_spec,
 )
-from ansible_collections.community.general.plugins.module_utils.database import SQLParseError, pg_quote_identifier
+from ansible_collections.community.general.plugins.module_utils.database import (
+    check_input,
+    pg_quote_identifier,
+    SQLParseError,
+)
 from ansible.module_utils._text import to_native
 
 executed_queries = []
@@ -151,9 +160,8 @@ class NotSupportedError(Exception):
 #
 
 def set_owner(cursor, schema, owner):
-    query = "ALTER SCHEMA %s OWNER TO %s" % (
-            pg_quote_identifier(schema, 'schema'),
-            pg_quote_identifier(owner, 'role'))
+    query = 'ALTER SCHEMA %s OWNER TO "%s"' % (
+            pg_quote_identifier(schema, 'schema'), owner)
     cursor.execute(query)
     executed_queries.append(query)
     return True
@@ -190,7 +198,7 @@ def schema_create(cursor, schema, owner):
     if not schema_exists(cursor, schema):
         query_fragments = ['CREATE SCHEMA %s' % pg_quote_identifier(schema, 'schema')]
         if owner:
-            query_fragments.append('AUTHORIZATION %s' % pg_quote_identifier(owner, 'role'))
+            query_fragments.append('AUTHORIZATION "%s"' % owner)
         query = ' '.join(query_fragments)
         cursor.execute(query)
         executed_queries.append(query)
@@ -227,6 +235,7 @@ def main():
         cascade_drop=dict(type="bool", default=False),
         state=dict(type="str", default="present", choices=["absent", "present"]),
         session_role=dict(type="str"),
+        trust_input=dict(type="bool", default=True),
     )
 
     module = AnsibleModule(
@@ -238,6 +247,13 @@ def main():
     owner = module.params["owner"]
     state = module.params["state"]
     cascade_drop = module.params["cascade_drop"]
+    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, schema, owner, session_role)
+
     changed = False
 
     conn_params = get_conn_params(module, module.params)

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

@@ -2,5 +2,6 @@
 db_name: 'ansible_db'
 db_user1: 'ansible_db_user1'
 db_user2: 'ansible_db_user2'
+dangerous_name: 'curious.anonymous"; SELECT * FROM information_schema.tables; --'
 db_session_role1: 'session_role1'
 db_session_role2: 'session_role2'

tests/integration/targets/postgresql_schema/tasks/postgresql_schema_initial.yml

@@ -61,6 +61,7 @@
     database: "{{ db_name }}"
     name: acme
     login_user: "{{ pg_user }}"
+    trust_input: yes
   register: result
 
 # Checks
@@ -144,6 +145,25 @@
     that:
       - result.rowcount == 0
 
+# Test: trust_input parameter
+- name: Create a new schema with potentially dangerous owner name
+  become_user: "{{ pg_user }}"
+  become: yes
+  postgresql_schema:
+    database: "{{ db_name }}"
+    name: acme
+    login_user: "{{ pg_user }}"
+    owner: "{{ dangerous_name }}"
+    trust_input: no
+  register: result
+  ignore_errors: yes
+
+# Checks
+- assert:
+    that:
+      - result is failed
+      - result.msg == 'Passed input \'{{ dangerous_name }}\' is potentially dangerous'
+
 # Test: CREATE SCHEMA; WITH TABLE for DROP CASCADE test
 - name: Create a new schema "acme"
   become_user: "{{ pg_user }}"