postgresql_user: add trust_input parameter (#116)

* postgresql: add input checks for potentially dangerous substrings

* postgresql_user: add trust_input parameter

* add CI, add changelog fragment

* fix CI

* moved input patterns outside is_input_dangerous function

* Update plugins/module_utils/database.py

Co-Authored-By: Thomas O'Donnell <andytom@users.noreply.github.com>

* Update plugins/module_utils/database.py

Co-Authored-By: Thomas O'Donnell <andytom@users.noreply.github.com>

* fix

Co-authored-by: Thomas O'Donnell <andytom@users.noreply.github.com>

changelogs/fragments/116-postgresql_user_add_trust_input_parameter.yml

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

plugins/module_utils/database.py

@@ -26,6 +26,21 @@
 # LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE
 # USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 
+import re
+
+
+# Input patterns for is_input_dangerous function:
+#
+# 1. '"' in string and '--' in string or
+# "'" in string and '--' in string
+PATTERN_1 = re.compile(r'(\'|\").*--')
+
+# 2. union \ intersect \ except + select
+PATTERN_2 = re.compile(r'(UNION|INTERSECT|EXCEPT).*SELECT', re.IGNORECASE)
+
+# 3. ';' and any KEY_WORDS
+PATTERN_3 = re.compile(r';.*(SELECT|UPDATE|INSERT|DELETE|DROP|TRUNCATE|ALTER)', re.IGNORECASE)
+
 
 class SQLParseError(Exception):
     pass
@@ -140,3 +155,47 @@ def mysql_quote_identifier(identifier, id_type):
             special_cased_fragments.append(fragment)
 
     return '.'.join(special_cased_fragments)
+
+
+def is_input_dangerous(string):
+    """Check if the passed string is potentially dangerous.
+    Can be used to prevent SQL injections.
+
+    Note: use this function only when you can't use
+      psycopg2's cursor.execute method parametrized
+      (typically with DDL queries).
+    """
+    if not string:
+        return False
+
+    for pattern in (PATTERN_1, PATTERN_2, PATTERN_3):
+        if re.search(pattern, string):
+            return True
+
+    return False
+
+
+def check_input(module, *args):
+    """Wrapper for is_input_dangerous function."""
+    needs_to_check = args
+
+    dangerous_elements = []
+
+    for elem in needs_to_check:
+        if isinstance(elem, str):
+            if is_input_dangerous(elem):
+                dangerous_elements.append(elem)
+
+        elif isinstance(elem, list):
+            for e in elem:
+                if is_input_dangerous(e):
+                    dangerous_elements.append(e)
+
+        else:
+            elem = str(elem)
+            if is_input_dangerous(elem):
+                dangerous_elements.append(elem)
+
+    if dangerous_elements:
+        module.fail_json(msg="Passed input '%s' is "
+                             "potentially dangerous" % ', '.join(dangerous_elements))

plugins/modules/database/postgresql/postgresql_user.py

@@ -147,6 +147,11 @@ options:
     description:
     - Add a comment on the user (equal to the COMMENT ON ROLE statement result).
     type: str
+  trust_input:
+    description:
+    - If C(no), check whether values of some parameters are potentially dangerous.
+    type: bool
+    default: yes
 notes:
 - The module creates a user (role) with login privilege by default.
   Use NOLOGIN role_attr_flags to change this behaviour.
@@ -252,7 +257,11 @@ except ImportError:
     pass
 
 from ansible.module_utils.basic import AnsibleModule
-from ansible_collections.community.general.plugins.module_utils.database import pg_quote_identifier, SQLParseError
+from ansible_collections.community.general.plugins.module_utils.database import (
+    pg_quote_identifier,
+    SQLParseError,
+    check_input,
+)
 from ansible_collections.community.general.plugins.module_utils.postgres import (
     connect_to_db,
     get_conn_params,
@@ -812,6 +821,7 @@ def main():
         session_role=dict(type='str'),
         groups=dict(type='list', elements='str'),
         comment=dict(type='str', default=None),
+        trust_input=dict(type='bool', default=True),
     )
     module = AnsibleModule(
         argument_spec=argument_spec,
@@ -838,6 +848,12 @@ def main():
         groups = [e.strip() for e in groups]
     comment = module.params["comment"]
 
+    trust_input = module.params['trust_input']
+    if not trust_input:
+        # Check input for potentially dangerous elements:
+        check_input(module, user, password, privs, expires,
+                    role_attr_flags, groups, comment)
+
     conn_params = get_conn_params(module, module.params, warn_db_default=False)
     db_connection = connect_to_db(module, conn_params)
     cursor = db_connection.cursor(cursor_factory=DictCursor)

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

@@ -1,3 +1,4 @@
 db_name: 'ansible_db'
 db_user1: 'ansible_db_user1'
 db_user2: 'ansible_db_user2'
+dangerous_name: 'curious.anonymous"; SELECT * FROM information_schema.tables; --'

tests/integration/targets/postgresql_user/tasks/postgresql_user_general.yml

@@ -717,6 +717,32 @@
       that:
       - result.rowcount == 2
 
+  ########################
+  # Test trust_input param
+
+  - name: Create role with potentially dangerous name, don't trust
+    <<: *task_parameters
+    postgresql_user:
+      <<: *pg_parameters
+      name: '{{ dangerous_name }}'
+      trust_input: no
+    ignore_errors: yes
+
+  - assert:
+      that:
+      - result is failed
+      - result.msg == 'Passed input \'{{ dangerous_name }}\' is potentially dangerous'
+
+  - name: Create role with potentially dangerous name, trust
+    <<: *task_parameters
+    postgresql_user:
+      <<: *pg_parameters
+      name: '{{ dangerous_name }}'
+
+  - assert:
+      that:
+      - result is changed
+
   always:
   #
   # Clean up
@@ -739,3 +765,4 @@
     - '{{ test_user2 }}'
     - '{{ test_group1 }}'
     - '{{ test_group2 }}'
+    - '{{ dangerous_name }}'

tests/unit/plugins/module_utils/test_database.py

@@ -1,6 +1,7 @@
 import pytest
 
 from ansible_collections.community.general.plugins.module_utils.database import (
+    is_input_dangerous,
     pg_quote_identifier,
     SQLParseError,
 )
@@ -76,6 +77,36 @@ HOW_MANY_DOTS = (
 VALID_QUOTES = ((test, VALID[test]) for test in sorted(VALID))
 INVALID_QUOTES = ((test[0], test[1], INVALID[test]) for test in sorted(INVALID))
 
+IS_STRINGS_DANGEROUS = (
+    (u'', False),
+    (u' ', False),
+    (u'alternative database', False),
+    (u'backup of TRUNCATED table', False),
+    (u'bob.dropper', False),
+    (u'd\'artagnan', False),
+    (u'user_with_select_update_truncate_right', False),
+    (u';DROP DATABASE fluffy_pets_photos', True),
+    (u';drop DATABASE fluffy_pets_photos', True),
+    (u'; TRUNCATE TABLE his_valuable_table', True),
+    (u'; truncate TABLE his_valuable_table', True),
+    (u'\'--', True),
+    (u'"--', True),
+    (u'\' union select username, password from admin_credentials', True),
+    (u'\' UNION SELECT username, password from admin_credentials', True),
+    (u'\' intersect select', True),
+    (u'\' INTERSECT select', True),
+    (u'\' except select', True),
+    (u'\' EXCEPT select', True),
+    (u';ALTER TABLE prices', True),
+    (u';alter table prices', True),
+    (u"; UPDATE products SET price = '0'", True),
+    (u";update products SET price = '0'", True),
+    (u"; DELETE FROM products", True),
+    (u"; delete FROM products", True),
+    (u"; SELECT * FROM products", True),
+    (u" ; select * from products", True),
+)
+
 
 @pytest.mark.parametrize("identifier, quoted_identifier", VALID_QUOTES)
 def test_valid_quotes(identifier, quoted_identifier):
@@ -98,3 +129,8 @@ def test_how_many_dots(identifier, id_type, quoted_identifier, msg):
         pg_quote_identifier('%s.more' % identifier, id_type)
 
     ex.match(msg)
+
+
+@pytest.mark.parametrize("string, result", IS_STRINGS_DANGEROUS)
+def test_is_input_dangerous(string, result):
+    assert is_input_dangerous(string) == result