perf: Avoid Oracle password modification SQL injection risks

apps/libs/ansible/modules/oracle_user.py

@@ -89,7 +89,12 @@ name:
     description: The name of the user to add or remove.
     returned: success
     type: str
+changed:
+    description: Whether the user was modified.
+    returned: success
+    type: bool
 '''
+import re
 
 from ansible.module_utils.basic import AnsibleModule
 
@@ -98,75 +103,172 @@ from libs.ansible.modules_utils.oracle_common import (
 )
 
 
+def validate_identifier(name):
+    """
+    Strictly validate Oracle identifiers (usernames, tablespace names)
+    Only letters, numbers, and underscores are allowed.
+    The length must be ≤ 30 characters, and the first character must be a letter.
+    """
+    if not name:
+        return False, "Identifier cannot be empty"
+    if len(name) > 30:
+        return False, "Identifier must be at most 30 characters"
+    if not re.match(r'^[A-Za-z][A-Za-z0-9_]*$', name):
+        msg = ("Identifier can only contain letters, numbers, "
+               "and underscores (must start with a letter)")
+        return False, msg
+    return True, ""
+
+
 def user_find(oracle_client, username):
-    user = None
+    user_find_sql = """
-    username = username.upper()
+        SELECT username, 
-    user_find_sql = "select username, " \
+               authentication_type, 
-                    "       authentication_type, " \
+               default_tablespace, 
-                    "       default_tablespace, " \
+               temporary_tablespace 
-                    "       temporary_tablespace " \
+        FROM dba_users 
-                    "from dba_users where username='%s'" % username
+        WHERE username = UPPER(:username)
-    rtn, err = oracle_client.execute(user_find_sql)
+    """
-    if isinstance(rtn, dict):
+    rtn, err = oracle_client.execute(user_find_sql, {'username': username})
-        user = rtn
+    if err:
-    return user
+        return None, err
+    if isinstance(rtn, list) and len(rtn) > 0:
+        return rtn[0], None
+    return rtn, None
+
+
+def get_identified_clause(auth_type, password):
+    auth_type = auth_type.lower()
+    if auth_type == 'external':
+        return "IDENTIFIED EXTERNALLY"
+    elif auth_type == 'global':
+        return "IDENTIFIED GLOBALLY"
+    elif auth_type == 'no_authentication':
+        return "IDENTIFIED BY VALUES ''"
+    elif auth_type == 'password':
+        if not password:
+            raise ValueError("Password is required for 'password' authentication type")
+        quote_password = password.replace('"', '""')
+        return f'IDENTIFIED BY "{quote_password}"'
+    else:
+        raise ValueError(f"Unsupported authentication type: {auth_type}")
 
 
 def user_add(
         module, oracle_client, username, password, auth_type,
-        default_tablespace, temporary_tablespace
+        default_tablespace, temporary_tablespace, update_password
 ):
-    username = username.upper()
+    valid, msg = validate_identifier(username)
-    extend_sql = None
+    if not valid:
-    user = user_find(oracle_client, username)
+        module.fail_json(msg=f"Invalid username: {msg}")
-    auth_type = auth_type.lower()
-    identified_suffix_map = {
-        'external': 'identified externally ',
-        'global': 'identified globally ',
-        'password': 'identified by "%s" ',
-    }
-    if user:
-        user_sql = "alter user %s " % username
-        user_sql += identified_suffix_map.get(auth_type, 'no authentication ') % password
 
-        if default_tablespace and default_tablespace.lower() != user['default_tablespace'].lower():
-            user_sql += 'default tablespace %s quota unlimited on %s ' % (default_tablespace, default_tablespace)
-        if temporary_tablespace and temporary_tablespace.lower() != user['temporary_tablespace'].lower():
-            user_sql += 'temporary tablespace %s ' % temporary_tablespace
-    else:
-        user_sql = "create user %s " % username
-        user_sql += identified_suffix_map.get(auth_type, 'no authentication ') % password
     if default_tablespace:
-            user_sql += 'default tablespace %s quota unlimited on %s ' % (default_tablespace, default_tablespace)
+        valid, msg = validate_identifier(default_tablespace)
-        if temporary_tablespace:
+        if not valid:
-            user_sql += 'temporary tablespace %s ' % temporary_tablespace
+            module.fail_json(msg=f"Invalid default tablespace: {msg}")
-        extend_sql = 'grant connect to %s' % username
+        default_tablespace = default_tablespace.upper()
 
-    rtn, err = oracle_client.execute(user_sql)
+    if temporary_tablespace:
+        valid, msg = validate_identifier(temporary_tablespace)
+        if not valid:
+            module.fail_json(msg=f"Invalid temporary tablespace: {msg}")
+        temporary_tablespace = temporary_tablespace.upper()
+
+    user, err = user_find(oracle_client, username)
     if err:
-        module.fail_json(msg='Cannot add/edit user %s: %s' % (username, err), changed=False)
+        module.fail_json(msg=f"Failed to check user existence: {err}")
+
+    desired_attrs = {
+        'auth_type': auth_type.lower(),
+        'default_tablespace': default_tablespace,
+        'temporary_tablespace': temporary_tablespace,
+    }
+    username = username.upper()
+    if user:
+        current_attrs = {
+            'auth_type': user['authentication_type'].lower(),
+            'default_tablespace': user['default_tablespace'],
+            'temporary_tablespace': user['temporary_tablespace']
+        }
+        need_change = False
+        if current_attrs['auth_type'] != desired_attrs['auth_type']:
+            need_change = True
+        if (desired_attrs['default_tablespace'] and
+                current_attrs['default_tablespace'] != desired_attrs['default_tablespace']):
+            need_change = True
+        if (desired_attrs['temporary_tablespace'] and
+                current_attrs['temporary_tablespace'] != desired_attrs['temporary_tablespace']):
+            need_change = True
+        if desired_attrs['auth_type'] == 'password' and update_password == 'always':
+            need_change = True
+        if not need_change:
+            module.exit_json(changed=False, name=username)
+
+        sql_parts = [f"ALTER USER {username}"]
+        identified_clause = get_identified_clause(auth_type, password)
+        sql_parts.append(identified_clause)
+
+        if (desired_attrs['default_tablespace'] and
+                current_attrs['default_tablespace'] != desired_attrs['default_tablespace']):
+            sql_parts.append(f"DEFAULT TABLESPACE {desired_attrs['default_tablespace']}")
+            sql_parts.append(f"QUOTA UNLIMITED ON {desired_attrs['default_tablespace']}")
+
+        if (desired_attrs['temporary_tablespace'] and
+                current_attrs['temporary_tablespace'] != desired_attrs['temporary_tablespace']):
+            sql_parts.append(f"TEMPORARY TABLESPACE {desired_attrs['temporary_tablespace']}")
+        user_sql = " ".join(sql_parts)
     else:
-        if extend_sql:
+        sql_parts = [f"CREATE USER {username}"]
-            oracle_client.execute(extend_sql)
+        identified_clause = get_identified_clause(auth_type, password)
-        module.exit_json(msg='User %s has been created.' % username, changed=True, name=username)
+        sql_parts.append(identified_clause)
+
+        if desired_attrs['default_tablespace']:
+            sql_parts.append(f"DEFAULT TABLESPACE {desired_attrs['default_tablespace']}")
+            sql_parts.append(f"QUOTA UNLIMITED ON {desired_attrs['default_tablespace']}")
+
+        if desired_attrs['temporary_tablespace']:
+            sql_parts.append(f"TEMPORARY TABLESPACE {desired_attrs['temporary_tablespace']}")
+        user_sql = " ".join(sql_parts)
+
+    try:
+        ret, err = oracle_client.execute(user_sql)
+        if err:
+            module.fail_json(msg=f"Failed to modify user {username}: {err}", changed=False)
+        oracle_client.commit()
+    except Exception as e:
+        module.fail_json(msg=f"Database error while modifying user {username}: {str(e)}", changed=False)
+        
+    try:
+        ret, err = oracle_client.execute(f'GRANT CREATE SESSION TO {username}')
+        if err:
+            module.fail_json(msg=f"Failed to grant create session to {username}: {err}", changed=False)
+        action = 'updated' if user else 'created'
+        module.exit_json(changed=True, name=username, msg=f"User {username} {action} successfully")
+    except Exception as e:
+        module.fail_json(msg=f"Database error while modifying user {username}: {str(e)}", changed=False)
 
 
 def user_remove(module, oracle_client, username):
-    user = user_find(oracle_client, username)
+    black_list = ['sys','system','dbsnmp']
+    if username.lower() in black_list:
+       module.fail_json(msg=f'Trying to drop an internal user: %s. Not allowed' % username)
 
-    if user:
+    user, err = user_find(oracle_client, username)
-        rtn, err = oracle_client.execute('drop user %s cascade' % username)
     if err:
-            module.fail_json(msg='Cannot drop user %s: %s' % (username, err), changed=False)
+        module.fail_json(msg=f"Failed to check user existence: {err}")
-        else:
-            module.exit_json(msg='User %s dropped.' % username, changed=True, name=username)
-    else:
-        module.exit_json(msg="User %s doesn't exist." % username, changed=False, name=username)
 
+    if not user:
+        module.exit_json(changed=False, name=username, msg=f"User {username} does not exist")
+
+    drop_sql = f"DROP USER {username.upper()} CASCADE"
+    try:
+        _, err = oracle_client.execute(drop_sql)
+        if err:
+            module.fail_json(msg=f"Failed to drop user {username}: {err}", changed=False)
+        module.exit_json(changed=True, name=username, msg=f"User {username} dropped successfully")
+    except Exception as e:
+        module.fail_json(msg=f"Database error while dropping user {username}: {str(e)}", changed=False)
 
-# =========================================
-# Module execution.
-#
 
 def main():
     argument_spec = oracle_common_argument_spec()
@@ -184,7 +286,7 @@ def main():
     )
     module = AnsibleModule(
         argument_spec=argument_spec,
-        supports_check_mode=True,
+        supports_check_mode=False,
     )
 
     authentication_type = module.params['authentication_type'] or 'password'
@@ -195,16 +297,21 @@ def main():
     update_password = module.params['update_password']
     temporary_tablespace = module.params['temporary_tablespace']
 
+    try:
         oracle_client = OracleClient(module)
+    except Exception as e:
+        module.fail_json(msg=f"Failed to connect to Oracle: {str(e)}")
+        return
+
     if state == 'present':
-        if password is None and update_password == 'always':
+        if not password and update_password == 'always':
             module.fail_json(
                 msg='password parameter required when adding a user unless update_password is set to on_create'
             )
         user_add(
             module, oracle_client, username=user, password=password,
             auth_type=authentication_type, default_tablespace=default_tablespace,
-            temporary_tablespace=temporary_tablespace
+            temporary_tablespace=temporary_tablespace, update_password=update_password
         )
     elif state == 'absent':
         user_remove(module, oracle_client, user)

apps/libs/ansible/modules_utils/oracle_common.py

@@ -65,11 +65,11 @@ class OracleClient(object):
                 )
         return self._cursor
 
-    def execute(self, sql, exception_to_fail=False):
+    def execute(self, sql, params=None, exception_to_fail=False):
         sql = sql[:-1] if sql.endswith(';') else sql
         result, error = None, None
         try:
-            self.cursor.execute(sql)
+            self.cursor.execute(sql, params)
             sql_header = self.cursor.description or []
             column_names = [description[0].lower() for description in sql_header]
             if column_names:
@@ -83,6 +83,11 @@ class OracleClient(object):
             self.module.fail_json(msg='Cannot execute sql: %s' % to_native(error))
         return result, error
 
+    def commit(self):
+        if self._conn is None:
+            raise DatabaseError('Cannot connect to database')
+        self._conn.commit()
+
     def close(self):
         try:
             if self._cursor: