diff --git a/changelogs/fragments/20221021-ec2_instance-tower_callback.yml b/changelogs/fragments/20221021-ec2_instance-tower_callback.yml
new file mode 100644
index 00000000000..ae961144f29
--- /dev/null
+++ b/changelogs/fragments/20221021-ec2_instance-tower_callback.yml
@@ -0,0 +1,5 @@
+minor_changes:
+- ec2_instance - refacter ``tower_callback`` code to handle parameter validation as part of the argument specification (https://github.com/ansible-collections/amazon.aws/pull/1199).
+- ec2_instance - the ``tower_callback`` parameter has been renamed to ``aap_callback``, ``tower_callback`` remains as an alias.  This change should have no observable effect for users outside the module documentation (https://github.com/ansible-collections/amazon.aws/pull/1199).
+security_fixes:
+- ec2_instance - fixes leak of password into logs when using ``tower_callback.windows=True`` and ``tower_callback.set_password`` (https://github.com/ansible-collections/amazon.aws/pull/1199).
diff --git a/plugins/module_utils/tower.py b/plugins/module_utils/tower.py
new file mode 100644
index 00000000000..dd7d9738a5e
--- /dev/null
+++ b/plugins/module_utils/tower.py
@@ -0,0 +1,83 @@
+# Copyright: Ansible Project
+# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)
+
+from __future__ import absolute_import, division, print_function
+__metaclass__ = type
+
+import string
+import textwrap
+
+from ansible.module_utils._text import to_native
+from ansible.module_utils.six.moves.urllib import parse as urlparse
+
+
+def _windows_callback_script(passwd=None):
+    script_url = 'https://raw.githubusercontent.com/ansible/ansible/devel/examples/scripts/ConfigureRemotingForAnsible.ps1'
+    if passwd is not None:
+        passwd = passwd.replace("'", "''")
+        script_tpl = """\
+        <powershell>
+        $admin = [adsi]('WinNT://./administrator, user')
+        $admin.PSBase.Invoke('SetPassword', '${PASS}')
+        Invoke-Expression ((New-Object System.Net.Webclient).DownloadString('${SCRIPT}'))
+        </powershell>
+        """
+    else:
+        script_tpl = """\
+        <powershell>
+        $admin = [adsi]('WinNT://./administrator, user')
+        Invoke-Expression ((New-Object System.Net.Webclient).DownloadString('${SCRIPT}'))
+        </powershell>
+        """
+
+    tpl = string.Template(textwrap.dedent(script_tpl))
+    return tpl.safe_substitute(PASS=passwd, SCRIPT=script_url)
+
+
+def _linux_callback_script(tower_address, template_id, host_config_key):
+    template_id = urlparse.quote(template_id)
+    tower_address = urlparse.quote(tower_address)
+    host_config_key = host_config_key.replace("'", "'\"'\"'")
+
+    script_tpl = """\
+    #!/bin/bash
+    set -x
+
+    retry_attempts=10
+    attempt=0
+    while [[ $attempt -lt $retry_attempts ]]
+    do
+      status_code=$(curl --max-time 10 -v -k -s -i \
+        --data 'host_config_key=${host_config_key}' \
+        'https://${tower_address}/api/v2/job_templates/${template_id}/callback/' \
+        | head -n 1 \
+        | awk '{print $2}')
+      if [[ $status_code == 404 ]]
+        then
+        status_code=$(curl --max-time 10 -v -k -s -i \
+          --data 'host_config_key=${host_config_key}' \
+          'https://${tower_address}/api/v1/job_templates/${template_id}/callback/' \
+          | head -n 1 \
+          | awk '{print $2}')
+        # fall back to using V1 API for Tower 3.1 and below, since v2 API will always 404
+      fi
+      if [[ $status_code == 201 ]]
+        then
+        exit 0
+      fi
+      attempt=$(( attempt + 1 ))
+      echo "$${status_code} received... retrying in 1 minute. (Attempt $${attempt})"
+      sleep 60
+    done
+    exit 1
+    """
+    tpl = string.Template(textwrap.dedent(script_tpl))
+    return tpl.safe_substitute(tower_address=tower_address,
+                               template_id=template_id,
+                               host_config_key=host_config_key)
+
+
+def tower_callback_script(tower_address, job_template_id, host_config_key, windows, passwd):
+    if windows:
+        return to_native(_windows_callback_script(passwd=passwd))
+    return _linux_callback_script(tower_address, job_template_id, host_config_key)
diff --git a/plugins/modules/ec2_instance.py b/plugins/modules/ec2_instance.py
index bd431042f09..2bb4a34b202 100644
--- a/plugins/modules/ec2_instance.py
+++ b/plugins/modules/ec2_instance.py
@@ -75,25 +75,44 @@
     description:
       - Opaque blob of data which is made available to the ec2 instance
     type: str
-  tower_callback:
+  aap_callback:
     description:
-      - Preconfigured user-data to enable an instance to perform a Tower callback (Linux only).
+      - Preconfigured user-data to enable an instance to perform an Ansible Automation Platform
+        callback (Linux only).
+      - For Windows instances, to enable remote access via Ansible set I(windows) to C(true), and
+        optionally set an admin password.
+      - If using I(windows) and I(set_password), callback ton Ansible Automation Platform will not
+        be performed but the instance will be ready to receive winrm connections from Ansible.
       - Mutually exclusive with I(user_data).
-      - For Windows instances, to enable remote access via Ansible set I(tower_callback.windows) to true, and optionally set an admin password.
-      - If using 'windows' and 'set_password', callback to Tower will not be performed but the instance will be ready to receive winrm connections from Ansible.
     type: dict
+    aliases: ['tower_callback']
     suboptions:
+      windows:
+        description:
+          - Set I(windows=True) to use powershell instead of bash for the callback script.
+        type: bool
+        default: False
+      set_password:
+        description:
+          - Optional admin password to use if I(windows=True).
+        type: str
       tower_address:
         description:
-        - IP address or DNS name of Tower server. Must be accessible via this address from the VPC that this instance will be launched in.
+          - IP address or DNS name of Tower server. Must be accessible via this address from the
+            VPC that this instance will be launched in.
+          - Required if I(windows=False).
         type: str
       job_template_id:
         description:
-        - Either the integer ID of the Tower Job Template, or the name (name supported only for Tower 3.2+).
+          - Either the integer ID of the Tower Job Template, or the name.
+            Using a name for the job template is not supported by Ansible Tower prior to version
+            3.2.
+          - Required if I(windows=False).
         type: str
       host_config_key:
         description:
-        - Host configuration secret key generated by the Tower job template.
+          - Host configuration secret key generated by the Tower job template.
+          - Required if I(windows=False).
         type: str
   tags:
     description:
@@ -915,71 +934,11 @@
 from ansible_collections.amazon.aws.plugins.module_utils.ec2 import get_ec2_security_group_ids_from_names
 from ansible_collections.amazon.aws.plugins.module_utils.tagging import boto3_tag_list_to_ansible_dict
 from ansible_collections.amazon.aws.plugins.module_utils.tagging import boto3_tag_specifications
+from ansible_collections.amazon.aws.plugins.module_utils.tower import tower_callback_script
 
 module = None
 
 
-def tower_callback_script(tower_conf, windows=False, passwd=None):
-    script_url = 'https://raw.githubusercontent.com/ansible/ansible/devel/examples/scripts/ConfigureRemotingForAnsible.ps1'
-    if windows and passwd is not None:
-        script_tpl = """<powershell>
-        $admin = [adsi]("WinNT://./administrator, user")
-        $admin.PSBase.Invoke("SetPassword", "{PASS}")
-        Invoke-Expression ((New-Object System.Net.Webclient).DownloadString('{SCRIPT}'))
-        </powershell>
-        """
-        return to_native(textwrap.dedent(script_tpl).format(PASS=passwd, SCRIPT=script_url))
-    elif windows and passwd is None:
-        script_tpl = """<powershell>
-        $admin = [adsi]("WinNT://./administrator, user")
-        Invoke-Expression ((New-Object System.Net.Webclient).DownloadString('{SCRIPT}'))
-        </powershell>
-        """
-        return to_native(textwrap.dedent(script_tpl).format(PASS=passwd, SCRIPT=script_url))
-    elif not windows:
-        for p in ['tower_address', 'job_template_id', 'host_config_key']:
-            if p not in tower_conf:
-                module.fail_json(msg="Incomplete tower_callback configuration. tower_callback.{0} not set.".format(p))
-
-        if isinstance(tower_conf['job_template_id'], string_types):
-            tower_conf['job_template_id'] = urlparse.quote(tower_conf['job_template_id'])
-        tpl = string.Template(textwrap.dedent("""#!/bin/bash
-        set -x
-
-        retry_attempts=10
-        attempt=0
-        while [[ $attempt -lt $retry_attempts ]]
-        do
-          status_code=`curl --max-time 10 -v -k -s -i \
-            --data "host_config_key=${host_config_key}" \
-            'https://${tower_address}/api/v2/job_templates/${template_id}/callback/' \
-            | head -n 1 \
-            | awk '{print $2}'`
-          if [[ $status_code == 404 ]]
-            then
-            status_code=`curl --max-time 10 -v -k -s -i \
-              --data "host_config_key=${host_config_key}" \
-              'https://${tower_address}/api/v1/job_templates/${template_id}/callback/' \
-              | head -n 1 \
-              | awk '{print $2}'`
-            # fall back to using V1 API for Tower 3.1 and below, since v2 API will always 404
-          fi
-          if [[ $status_code == 201 ]]
-            then
-            exit 0
-          fi
-          attempt=$(( attempt + 1 ))
-          echo "$${status_code} received... retrying in 1 minute. (Attempt $${attempt})"
-          sleep 60
-        done
-        exit 1
-        """))
-        return tpl.safe_substitute(tower_address=tower_conf['tower_address'],
-                                   template_id=tower_conf['job_template_id'],
-                                   host_config_key=tower_conf['host_config_key'])
-    raise NotImplementedError("Only windows with remote-prep or non-windows with tower job callback supported so far.")
-
-
 def build_volume_spec(params):
     volumes = params.get('volumes') or []
     for volume in volumes:
@@ -1210,6 +1169,21 @@ def discover_security_groups(group, groups, parent_vpc_id=None, subnet_id=None):
     return []
 
 
+def build_userdata(params):
+    if params.get('user_data') is not None:
+        return {'UserData': to_native(params.get('user_data'))}
+    if params.get('aap_callback'):
+        userdata = tower_callback_script(
+            tower_address=params.get('aap_callback').get('tower_address'),
+            job_template_id=params.get('aap_callback').get('job_template_id'),
+            host_config_key=params.get('aap_callback').get('host_config_key'),
+            windows=params.get('aap_callback').get('windows'),
+            passwd=params.get('aap_callback').get('set_passwd'),
+        )
+        return {'UserData': userdata}
+    return {}
+
+
 def build_top_level_options(params):
     spec = {}
     if params.get('image_id'):
@@ -1226,14 +1200,8 @@ def build_top_level_options(params):
 
     if params.get('key_name') is not None:
         spec['KeyName'] = params.get('key_name')
-    if params.get('user_data') is not None:
-        spec['UserData'] = to_native(params.get('user_data'))
-    elif params.get('tower_callback') is not None:
-        spec['UserData'] = tower_callback_script(
-            tower_conf=params.get('tower_callback'),
-            windows=params.get('tower_callback').get('windows', False),
-            passwd=params.get('tower_callback').get('set_password'),
-        )
+
+    spec.update(build_userdata(params))
 
     if params.get('launch_template') is not None:
         spec['LaunchTemplate'] = {}
@@ -1931,6 +1899,7 @@ def build_filters():
 def main():
     global module
     global client
+
     argument_spec = dict(
         state=dict(default='present', choices=['present', 'started', 'running', 'stopped', 'restarted', 'rebooted', 'terminated', 'absent']),
         wait=dict(default=True, type='bool'),
@@ -1941,7 +1910,19 @@ def main():
         image_id=dict(type='str'),
         instance_type=dict(type='str'),
         user_data=dict(type='str'),
-        tower_callback=dict(type='dict'),
+        aap_callback=dict(
+            type='dict', aliases=['tower_callback'],
+            required_if=[
+                ('windows', False, ('tower_address', 'job_template_id', 'host_config_key',), False),
+            ],
+            options=dict(
+                windows=dict(type='bool', default=False),
+                set_password=dict(type='str', no_log=True),
+                tower_address=dict(type='str'),
+                job_template_id=dict(type='str'),
+                host_config_key=dict(type='str', no_log=True),
+            ),
+        ),
         ebs_optimized=dict(type='bool'),
         vpc_subnet_id=dict(type='str', aliases=['subnet_id']),
         availability_zone=dict(type='str'),
@@ -1978,7 +1959,7 @@ def main():
         mutually_exclusive=[
             ['security_groups', 'security_group'],
             ['availability_zone', 'vpc_subnet_id'],
-            ['tower_callback', 'user_data'],
+            ['aap_callback', 'user_data'],
             ['image_id', 'image'],
             ['exact_count', 'count'],
             ['exact_count', 'instance_ids'],
diff --git a/tests/unit/module_utils/test_tower.py b/tests/unit/module_utils/test_tower.py
new file mode 100644
index 00000000000..9e1d9021311
--- /dev/null
+++ b/tests/unit/module_utils/test_tower.py
@@ -0,0 +1,40 @@
+# (c) 2022 Red Hat Inc.
+#
+# This file is part of Ansible
+# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)
+
+from __future__ import (absolute_import, division, print_function)
+__metaclass__ = type
+
+# import pytest
+
+import ansible_collections.amazon.aws.plugins.module_utils.tower as utils_tower
+
+WINDOWS_DOWNLOAD = "Invoke-Expression ((New-Object System.Net.Webclient).DownloadString(" \
+    "'https://raw.githubusercontent.com/ansible/ansible/devel/examples/scripts/ConfigureRemotingForAnsible.ps1'))"
+EXAMPLE_PASSWORD = 'MY_EXAMPLE_PASSWORD'
+WINDOWS_INVOKE = "$admin.PSBase.Invoke('SetPassword', 'MY_EXAMPLE_PASSWORD'"
+
+EXAMPLE_TOWER = "tower.example.com"
+EXAMPLE_TEMPLATE = 'My Template'
+EXAMPLE_KEY = '123EXAMPLE123'
+LINUX_TRIGGER_V1 = 'https://tower.example.com/api/v1/job_templates/My%20Template/callback/'
+LINUX_TRIGGER_V2 = 'https://tower.example.com/api/v2/job_templates/My%20Template/callback/'
+
+
+def test_windows_callback_no_password():
+    user_data = utils_tower._windows_callback_script()
+    assert WINDOWS_DOWNLOAD in user_data
+    assert 'SetPassword' not in user_data
+
+
+def test_windows_callback_password():
+    user_data = utils_tower._windows_callback_script(EXAMPLE_PASSWORD)
+    assert WINDOWS_DOWNLOAD in user_data
+    assert WINDOWS_INVOKE in user_data
+
+
+def test_linux_callback_with_name():
+    user_data = utils_tower._linux_callback_script(EXAMPLE_TOWER, EXAMPLE_TEMPLATE, EXAMPLE_KEY)
+    assert LINUX_TRIGGER_V1 in user_data
+    assert LINUX_TRIGGER_V2 in user_data