postgresql_db: add trust_input parameter (#184)

plugins/modules/database/postgresql/postgresql_db.py

@@ -105,6 +105,11 @@ options:
       - Provides additional arguments when I(state) is C(dump).
       - Cannot be used with dump-file-format-related arguments like ``--format=d``.
     type: str
+  trust_input:
+    description:
+    - If C(no), check whether values of some parameters are potentially dangerous.
+    type: bool
+    default: yes
 seealso:
 - name: CREATE DATABASE reference
   description: Complete reference of the CREATE DATABASE command documentation.
@@ -207,7 +212,10 @@ else:
 
 import ansible_collections.community.general.plugins.module_utils.postgres as pgutils
 from ansible.module_utils.basic import AnsibleModule
-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,
+    SQLParseError,
+)
 from ansible.module_utils.six import iteritems
 from ansible.module_utils.six.moves import shlex_quote
 from ansible.module_utils._text import to_native
@@ -224,18 +232,14 @@ class NotSupportedError(Exception):
 
 
 def set_owner(cursor, db, owner):
-    query = 'ALTER DATABASE %s OWNER TO "%s"' % (
-            pg_quote_identifier(db, 'database'),
-            owner)
+    query = 'ALTER DATABASE "%s" OWNER TO "%s"' % (db, owner)
     executed_commands.append(query)
     cursor.execute(query)
     return True
 
 
 def set_conn_limit(cursor, db, conn_limit):
-    query = "ALTER DATABASE %s CONNECTION LIMIT %s" % (
-            pg_quote_identifier(db, 'database'),
-            conn_limit)
+    query = 'ALTER DATABASE "%s" CONNECTION LIMIT %s' % (db, conn_limit)
     executed_commands.append(query)
     cursor.execute(query)
     return True
@@ -270,7 +274,7 @@ def db_exists(cursor, db):
 
 def db_delete(cursor, db):
     if db_exists(cursor, db):
-        query = "DROP DATABASE %s" % pg_quote_identifier(db, 'database')
+        query = 'DROP DATABASE "%s"' % db
         executed_commands.append(query)
         cursor.execute(query)
         return True
@@ -281,11 +285,11 @@ def db_delete(cursor, db):
 def db_create(cursor, db, owner, template, encoding, lc_collate, lc_ctype, conn_limit, tablespace):
     params = dict(enc=encoding, collate=lc_collate, ctype=lc_ctype, conn_limit=conn_limit, tablespace=tablespace)
     if not db_exists(cursor, db):
-        query_fragments = ['CREATE DATABASE %s' % pg_quote_identifier(db, 'database')]
+        query_fragments = ['CREATE DATABASE "%s"' % db]
         if owner:
             query_fragments.append('OWNER "%s"' % owner)
         if template:
-            query_fragments.append('TEMPLATE %s' % pg_quote_identifier(template, 'database'))
+            query_fragments.append('TEMPLATE "%s"' % template)
         if encoding:
             query_fragments.append('ENCODING %(enc)s')
         if lc_collate:
@@ -293,7 +297,7 @@ def db_create(cursor, db, owner, template, encoding, lc_collate, lc_ctype, conn_
         if lc_ctype:
             query_fragments.append('LC_CTYPE %(ctype)s')
         if tablespace:
-            query_fragments.append('TABLESPACE %s' % pg_quote_identifier(tablespace, 'tablespace'))
+            query_fragments.append('TABLESPACE "%s"' % tablespace)
         if conn_limit:
             query_fragments.append("CONNECTION LIMIT %(conn_limit)s" % {"conn_limit": conn_limit})
         query = ' '.join(query_fragments)
@@ -491,9 +495,7 @@ def do_with_password(module, cmd, password):
 
 
 def set_tablespace(cursor, db, tablespace):
-    query = "ALTER DATABASE %s SET TABLESPACE %s" % (
-            pg_quote_identifier(db, 'database'),
-            pg_quote_identifier(tablespace, 'tablespace'))
+    query = 'ALTER DATABASE "%s" SET TABLESPACE "%s"' % (db, tablespace)
     executed_commands.append(query)
     cursor.execute(query)
     return True
@@ -520,6 +522,7 @@ def main():
         conn_limit=dict(type='str', default=''),
         tablespace=dict(type='path', default=''),
         dump_extra_args=dict(type='str', default=None),
+        trust_input=dict(type='bool', default=True),
     )
 
     module = AnsibleModule(
@@ -542,6 +545,12 @@ def main():
     conn_limit = module.params['conn_limit']
     tablespace = module.params['tablespace']
     dump_extra_args = module.params['dump_extra_args']
+    trust_input = module.params['trust_input']
+
+    # Check input
+    if not trust_input:
+        # Check input for potentially dangerous elements:
+        check_input(module, owner, conn_limit, encoding, db, template, tablespace, session_role)
 
     raw_connection = state in ("dump", "restore")
 

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

@@ -4,3 +4,8 @@ db_user2: 'ansible.db.user2'
 tmp_dir: '/tmp'
 db_session_role1: 'session_role1'
 db_session_role2: 'session_role2'
+
+# To test trust_input parameter and
+# possibility to create a database with dots in its name
+db_name_with_dot: 'db.name'
+suspicious_db_name: '{{ db_name_with_dot }}"; --'

tests/integration/targets/postgresql_db/tasks/state_dump_restore.yml

@@ -140,6 +140,77 @@
        - "'47' in result.stdout"
        - "'Joe Smith' in result.stdout"
 
+############################
+# 1. Test trust_input parameter
+# 2. Test db name containing dots
+
+- name: state dump/restore - create database, trust_input no
+  become: yes
+  become_user: "{{ pg_user }}"
+  postgresql_db:
+    state: present
+    name: "{{ suspicious_db_name }}"
+    owner: "{{ db_user1 }}"
+    login_user: "{{ pg_user }}"
+    trust_input: no
+  register: result
+  ignore_errors: yes
+
+- assert:
+    that:
+    - result is failed
+    - result.msg == 'Passed input \'{{ suspicious_db_name }}\' is potentially dangerous'
+
+- name: state dump/restore - create database, trust_input yes explicitly
+  become: yes
+  become_user: "{{ pg_user }}"
+  postgresql_db:
+    state: present
+    name: "{{ suspicious_db_name }}"
+    owner: "{{ db_user1 }}"
+    login_user: "{{ pg_user }}"
+    trust_input: yes
+  register: result
+
+- assert:
+    that:
+    - result is changed
+
+- name: test state=restore to restore the database (expect changed=true)
+  become: yes
+  become_user: "{{ pg_user }}"
+  postgresql_db:
+    name: "{{ db_name_with_dot }}"
+    target: "{{ db_file_name }}"
+    owner: "{{ db_user1 }}"
+    login_user: '{{(test_fixture == "user")|ternary(db_user1, pg_user)}}'
+    target_opts: '{{(test_fixture == "user")|ternary("-n public", omit)}}'
+    login_host:  '{{(test_fixture == "user")|ternary("localhost", omit)}}'
+    login_password: '{{(test_fixture == "user")|ternary("password", omit)}}'
+    state: restore
+  register: result
+
+- name: assert output message restore the database
+  assert:
+    that:
+    - result is changed
+
+- name: state dump/restore - remove databases
+  become: yes
+  become_user: "{{ pg_user }}"
+  postgresql_db:
+    state: absent
+    name: "{{ db_name_with_dot }}"
+    owner: "{{ db_user1 }}"
+    login_user: "{{ pg_user }}"
+    trust_input: yes
+  register: result
+
+- assert:
+    that:
+    - result is changed
+
+# Clean up
 - name: state dump/restore - remove database name
   postgresql_db:
     name: "{{ db_name }}"