Adding upstream version 9.4.0+dfsg.upstream/9.4.0+dfsg

Signed-off-by: Daniel Baumann <daniel.baumann@progress-linux.org>

7 files changed, 681 insertions, 674 deletions

diff --git a/ansible_collections/amazon/aws/plugins/lookup/aws_account_attribute.py b/ansible_collections/amazon/aws/plugins/lookup/aws_account_attribute.py
index 415b76d75..180c40f8f 100644
--- a/ansible_collections/amazon/aws/plugins/lookup/aws_account_attribute.py
+++ b/ansible_collections/amazon/aws/plugins/lookup/aws_account_attribute.py
@@ -1,16 +1,12 @@
+# -*- coding: utf-8 -*-
+
 # (c) 2017 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
 
-DOCUMENTATION = '''
+DOCUMENTATION = r"""
 name: aws_account_attribute
 author:
   - Sloane Hertel (@s-hertel) <shertel@redhat.com>
-extends_documentation_fragment:
-  - amazon.aws.boto3
-  - amazon.aws.aws_credentials
-  - amazon.aws.aws_region
 short_description: Look up AWS account attributes
 description:
   - Describes attributes of your AWS account. You can specify one of the listed
@@ -26,9 +22,13 @@ options:
       - max-elastic-ips
       - vpc-max-elastic-ips
       - has-ec2-classic
-'''
+extends_documentation_fragment:
+  - amazon.aws.boto3
+  - amazon.aws.common.plugins
+  - amazon.aws.region.plugins
+"""
 
-EXAMPLES = """
+EXAMPLES = r"""
 vars:
   has_ec2_classic: "{{ lookup('aws_account_attribute', attribute='has-ec2-classic') }}"
   # true | false
@@ -39,10 +39,9 @@ vars:
   account_details: "{{ lookup('aws_account_attribute', wantlist='true') }}"
   # {'default-vpc': ['vpc-xxxxxxxx'], 'max-elastic-ips': ['5'], 'max-instances': ['20'],
   #  'supported-platforms': ['VPC', 'EC2'], 'vpc-max-elastic-ips': ['5'], 'vpc-max-security-groups-per-interface': ['5']}
-
 """
 
-RETURN = """
+RETURN = r"""
 _raw:
   description:
     Returns a boolean when I(attribute) is check_ec2_classic. Otherwise returns the value(s) of the attribute
@@ -50,87 +49,50 @@ _raw:
 """
 
 try:
-    import boto3
     import botocore
 except ImportError:
-    pass  # will be captured by imported HAS_BOTO3
+    pass  # Handled by AWSLookupBase
 
 from ansible.errors import AnsibleLookupError
 from ansible.module_utils._text import to_native
-from ansible.module_utils.basic import missing_required_lib
-from ansible.plugins.lookup import LookupBase
-
-from ansible_collections.amazon.aws.plugins.module_utils.ec2 import AWSRetry
-from ansible_collections.amazon.aws.plugins.module_utils.ec2 import HAS_BOTO3
 
+from ansible_collections.amazon.aws.plugins.module_utils.retries import AWSRetry
+from ansible_collections.amazon.aws.plugins.plugin_utils.lookup import AWSLookupBase
 
-def _boto3_conn(region, credentials):
-    boto_profile = credentials.pop('aws_profile', None)
 
-    try:
-        connection = boto3.session.Session(profile_name=boto_profile).client('ec2', region, **credentials)
-    except (botocore.exceptions.ProfileNotFound, botocore.exceptions.PartialCredentialsError):
-        if boto_profile:
-            try:
-                connection = boto3.session.Session(profile_name=boto_profile).client('ec2', region)
-            except (botocore.exceptions.ProfileNotFound, botocore.exceptions.PartialCredentialsError):
-                raise AnsibleLookupError("Insufficient credentials found.")
-        else:
-            raise AnsibleLookupError("Insufficient credentials found.")
-    return connection
-
-
-def _get_credentials(options):
-    credentials = {}
-    credentials['aws_profile'] = options['aws_profile']
-    credentials['aws_secret_access_key'] = options['aws_secret_key']
-    credentials['aws_access_key_id'] = options['aws_access_key']
-    if options['aws_security_token']:
-        credentials['aws_session_token'] = options['aws_security_token']
-
-    return credentials
-
-
-@AWSRetry.jittered_backoff(retries=10)
 def _describe_account_attributes(client, **params):
-    return client.describe_account_attributes(**params)
+    return client.describe_account_attributes(aws_retry=True, **params)
 
 
-class LookupModule(LookupBase):
+class LookupModule(AWSLookupBase):
     def run(self, terms, variables, **kwargs):
+        super().run(terms, variables, **kwargs)
 
-        if not HAS_BOTO3:
-            raise AnsibleLookupError(missing_required_lib('botocore and boto3'))
-
-        self.set_options(var_options=variables, direct=kwargs)
-        boto_credentials = _get_credentials(self._options)
-
-        region = self._options['region']
-        client = _boto3_conn(region, boto_credentials)
+        client = self.client("ec2", AWSRetry.jittered_backoff())
 
-        attribute = kwargs.get('attribute')
-        params = {'AttributeNames': []}
+        attribute = kwargs.get("attribute")
+        params = {"AttributeNames": []}
         check_ec2_classic = False
-        if 'has-ec2-classic' == attribute:
+        if "has-ec2-classic" == attribute:
             check_ec2_classic = True
-            params['AttributeNames'] = ['supported-platforms']
+            params["AttributeNames"] = ["supported-platforms"]
         elif attribute:
-            params['AttributeNames'] = [attribute]
+            params["AttributeNames"] = [attribute]
 
         try:
-            response = _describe_account_attributes(client, **params)['AccountAttributes']
+            response = _describe_account_attributes(client, **params)["AccountAttributes"]
         except (botocore.exceptions.ClientError, botocore.exceptions.BotoCoreError) as e:
-            raise AnsibleLookupError("Failed to describe account attributes: %s" % to_native(e))
+            raise AnsibleLookupError(f"Failed to describe account attributes: {to_native(e)}")
 
         if check_ec2_classic:
             attr = response[0]
-            return any(value['AttributeValue'] == 'EC2' for value in attr['AttributeValues'])
+            return any(value["AttributeValue"] == "EC2" for value in attr["AttributeValues"])
 
         if attribute:
             attr = response[0]
-            return [value['AttributeValue'] for value in attr['AttributeValues']]
+            return [value["AttributeValue"] for value in attr["AttributeValues"]]
 
         flattened = {}
         for k_v_dict in response:
-            flattened[k_v_dict['AttributeName']] = [value['AttributeValue'] for value in k_v_dict['AttributeValues']]
+            flattened[k_v_dict["AttributeName"]] = [value["AttributeValue"] for value in k_v_dict["AttributeValues"]]
         return flattened
diff --git a/ansible_collections/amazon/aws/plugins/lookup/aws_collection_constants.py b/ansible_collections/amazon/aws/plugins/lookup/aws_collection_constants.py
new file mode 100644
index 000000000..35f05c94e
--- /dev/null
+++ b/ansible_collections/amazon/aws/plugins/lookup/aws_collection_constants.py
@@ -0,0 +1,82 @@
+# -*- coding: utf-8 -*-
+
+# (c) 2023 Ansible Project
+# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)
+
+DOCUMENTATION = r"""
+name: aws_collection_constants
+author:
+  - Mark Chappell (@tremble)
+short_description: expose various collection related constants
+version_added: 6.0.0
+description:
+  - Exposes various collection related constants for use in integration tests.
+options:
+  _terms:
+    description: Name of the constant.
+    choices:
+      - MINIMUM_BOTOCORE_VERSION
+      - MINIMUM_BOTO3_VERSION
+      - HAS_BOTO3
+      - AMAZON_AWS_COLLECTION_VERSION
+      - AMAZON_AWS_COLLECTION_NAME
+      - COMMUNITY_AWS_COLLECTION_VERSION
+      - COMMUNITY_AWS_COLLECTION_NAME
+    required: True
+"""
+
+EXAMPLES = r"""
+"""
+
+RETURN = r"""
+_raw:
+  description: value
+  type: str
+"""
+
+from ansible.errors import AnsibleLookupError
+from ansible.plugins.lookup import LookupBase
+
+import ansible_collections.amazon.aws.plugins.module_utils.botocore as botocore_utils
+import ansible_collections.amazon.aws.plugins.module_utils.common as common_utils
+
+try:
+    import ansible_collections.community.aws.plugins.module_utils.common as community_utils
+
+    HAS_COMMUNITY = True
+except ImportError:
+    HAS_COMMUNITY = False
+
+
+class LookupModule(LookupBase):
+    def lookup_constant(self, name):
+        if name == "MINIMUM_BOTOCORE_VERSION":
+            return botocore_utils.MINIMUM_BOTOCORE_VERSION
+        if name == "MINIMUM_BOTO3_VERSION":
+            return botocore_utils.MINIMUM_BOTO3_VERSION
+        if name == "HAS_BOTO3":
+            return botocore_utils.HAS_BOTO3
+
+        if name == "AMAZON_AWS_COLLECTION_VERSION":
+            return common_utils.AMAZON_AWS_COLLECTION_VERSION
+        if name == "AMAZON_AWS_COLLECTION_NAME":
+            return common_utils.AMAZON_AWS_COLLECTION_NAME
+
+        if name == "COMMUNITY_AWS_COLLECTION_VERSION":
+            if not HAS_COMMUNITY:
+                raise AnsibleLookupError("Unable to load ansible_collections.community.aws.plugins.module_utils.common")
+            return community_utils.COMMUNITY_AWS_COLLECTION_VERSION
+        if name == "COMMUNITY_AWS_COLLECTION_NAME":
+            if not HAS_COMMUNITY:
+                raise AnsibleLookupError("Unable to load ansible_collections.community.aws.plugins.module_utils.common")
+            return community_utils.COMMUNITY_AWS_COLLECTION_NAME
+
+    def run(self, terms, variables, **kwargs):
+        self.set_options(var_options=variables, direct=kwargs)
+        if not terms:
+            raise AnsibleLookupError("Constant name not provided")
+        if len(terms) > 1:
+            raise AnsibleLookupError("Multiple constant names provided")
+        name = terms[0].upper()
+
+        return [self.lookup_constant(name)]
diff --git a/ansible_collections/amazon/aws/plugins/lookup/aws_secret.py b/ansible_collections/amazon/aws/plugins/lookup/aws_secret.py
deleted file mode 100644
index 0f694cfa0..000000000
--- a/ansible_collections/amazon/aws/plugins/lookup/aws_secret.py
+++ /dev/null
@@ -1,295 +0,0 @@
-# Copyright: (c) 2018, Aaron Smith <ajsmith10381@gmail.com>
-# 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
-
-DOCUMENTATION = r'''
-name: aws_secret
-author:
-  - Aaron Smith (!UNKNOWN) <ajsmith10381@gmail.com>
-extends_documentation_fragment:
-  - amazon.aws.boto3
-  - amazon.aws.aws_credentials
-  - amazon.aws.aws_region
-
-short_description: Look up secrets stored in AWS Secrets Manager
-description:
-  - Look up secrets stored in AWS Secrets Manager provided the caller
-    has the appropriate permissions to read the secret.
-  - Lookup is based on the secret's I(Name) value.
-  - Optional parameters can be passed into this lookup; I(version_id) and I(version_stage)
-options:
-  _terms:
-    description: Name of the secret to look up in AWS Secrets Manager.
-    required: True
-  bypath:
-    description: A boolean to indicate whether the parameter is provided as a hierarchy.
-    default: false
-    type: boolean
-    version_added: 1.4.0
-  nested:
-    description: A boolean to indicate the secret contains nested values.
-    type: boolean
-    default: false
-    version_added: 1.4.0
-  version_id:
-    description: Version of the secret(s).
-    required: False
-  version_stage:
-    description: Stage of the secret version.
-    required: False
-  join:
-    description:
-      - Join two or more entries to form an extended secret.
-      - This is useful for overcoming the 4096 character limit imposed by AWS.
-      - No effect when used with I(bypath).
-    type: boolean
-    default: false
-  on_deleted:
-    description:
-      - Action to take if the secret has been marked for deletion.
-      - C(error) will raise a fatal error when the secret has been marked for deletion.
-      - C(skip) will silently ignore the deleted secret.
-      - C(warn) will skip over the deleted secret but issue a warning.
-    default: error
-    type: string
-    choices: ['error', 'skip', 'warn']
-    version_added: 2.0.0
-  on_missing:
-    description:
-      - Action to take if the secret is missing.
-      - C(error) will raise a fatal error when the secret is missing.
-      - C(skip) will silently ignore the missing secret.
-      - C(warn) will skip over the missing secret but issue a warning.
-    default: error
-    type: string
-    choices: ['error', 'skip', 'warn']
-  on_denied:
-    description:
-      - Action to take if access to the secret is denied.
-      - C(error) will raise a fatal error when access to the secret is denied.
-      - C(skip) will silently ignore the denied secret.
-      - C(warn) will skip over the denied secret but issue a warning.
-    default: error
-    type: string
-    choices: ['error', 'skip', 'warn']
-'''
-
-EXAMPLES = r"""
- - name: lookup secretsmanager secret in the current region
-   debug: msg="{{ lookup('amazon.aws.aws_secret', '/path/to/secrets', bypath=true) }}"
-
- - name: Create RDS instance with aws_secret lookup for password param
-   rds:
-     command: create
-     instance_name: app-db
-     db_engine: MySQL
-     size: 10
-     instance_type: db.m1.small
-     username: dbadmin
-     password: "{{ lookup('amazon.aws.aws_secret', 'DbSecret') }}"
-     tags:
-       Environment: staging
-
- - name: skip if secret does not exist
-   debug: msg="{{ lookup('amazon.aws.aws_secret', 'secret-not-exist', on_missing='skip')}}"
-
- - name: warn if access to the secret is denied
-   debug: msg="{{ lookup('amazon.aws.aws_secret', 'secret-denied', on_denied='warn')}}"
-
- - name: lookup secretsmanager secret in the current region using the nested feature
-   debug: msg="{{ lookup('amazon.aws.aws_secret', 'secrets.environments.production.password', nested=true) }}"
-   # The secret can be queried using the following syntax: `aws_secret_object_name.key1.key2.key3`.
-   # If an object is of the form `{"key1":{"key2":{"key3":1}}}` the query would return the value `1`.
- - name: lookup secretsmanager secret in a specific region using specified region and aws profile using nested feature
-   debug: >
-    msg="{{ lookup('amazon.aws.aws_secret', 'secrets.environments.production.password', region=region, aws_profile=aws_profile,
-    aws_access_key=aws_access_key, aws_secret_key=aws_secret_key, nested=true) }}"
-   # The secret can be queried using the following syntax: `aws_secret_object_name.key1.key2.key3`.
-   # If an object is of the form `{"key1":{"key2":{"key3":1}}}` the query would return the value `1`.
-   # Region is the AWS region where the AWS secret is stored.
-   # AWS_profile is the aws profile to use, that has access to the AWS secret.
-"""
-
-RETURN = r"""
-_raw:
-  description:
-    Returns the value of the secret stored in AWS Secrets Manager.
-"""
-
-import json
-
-try:
-    import boto3
-    import botocore
-except ImportError:
-    pass  # will be captured by imported HAS_BOTO3
-
-from ansible.errors import AnsibleLookupError
-from ansible.module_utils.six import string_types
-from ansible.module_utils._text import to_native
-from ansible.module_utils.basic import missing_required_lib
-from ansible.plugins.lookup import LookupBase
-
-from ansible_collections.amazon.aws.plugins.module_utils.core import is_boto3_error_code
-from ansible_collections.amazon.aws.plugins.module_utils.core import is_boto3_error_message
-from ansible_collections.amazon.aws.plugins.module_utils.ec2 import HAS_BOTO3
-
-
-def _boto3_conn(region, credentials):
-    boto_profile = credentials.pop('aws_profile', None)
-
-    try:
-        connection = boto3.session.Session(profile_name=boto_profile).client('secretsmanager', region, **credentials)
-    except (botocore.exceptions.ProfileNotFound, botocore.exceptions.PartialCredentialsError):
-        if boto_profile:
-            try:
-                connection = boto3.session.Session(profile_name=boto_profile).client('secretsmanager', region)
-            except (botocore.exceptions.ProfileNotFound, botocore.exceptions.PartialCredentialsError):
-                raise AnsibleLookupError("Insufficient credentials found.")
-        else:
-            raise AnsibleLookupError("Insufficient credentials found.")
-    return connection
-
-
-class LookupModule(LookupBase):
-    def run(self, terms, variables=None, boto_profile=None, aws_profile=None,
-            aws_secret_key=None, aws_access_key=None, aws_security_token=None, region=None,
-            bypath=False, nested=False, join=False, version_stage=None, version_id=None, on_missing='error',
-            on_denied='error', on_deleted='error'):
-        '''
-                   :arg terms: a list of lookups to run.
-                       e.g. ['parameter_name', 'parameter_name_too' ]
-                   :kwarg variables: ansible variables active at the time of the lookup
-                   :kwarg aws_secret_key: identity of the AWS key to use
-                   :kwarg aws_access_key: AWS secret key (matching identity)
-                   :kwarg aws_security_token: AWS session key if using STS
-                   :kwarg decrypt: Set to True to get decrypted parameters
-                   :kwarg region: AWS region in which to do the lookup
-                   :kwarg bypath: Set to True to do a lookup of variables under a path
-                   :kwarg nested: Set to True to do a lookup of nested secrets
-                   :kwarg join: Join two or more entries to form an extended secret
-                   :kwarg version_stage: Stage of the secret version
-                   :kwarg version_id: Version of the secret(s)
-                   :kwarg on_missing: Action to take if the secret is missing
-                   :kwarg on_deleted: Action to take if the secret is marked for deletion
-                   :kwarg on_denied: Action to take if access to the secret is denied
-                   :returns: A list of parameter values or a list of dictionaries if bypath=True.
-               '''
-        if not HAS_BOTO3:
-            raise AnsibleLookupError(missing_required_lib('botocore and boto3'))
-
-        deleted = on_deleted.lower()
-        if not isinstance(deleted, string_types) or deleted not in ['error', 'warn', 'skip']:
-            raise AnsibleLookupError('"on_deleted" must be a string and one of "error", "warn" or "skip", not %s' % deleted)
-
-        missing = on_missing.lower()
-        if not isinstance(missing, string_types) or missing not in ['error', 'warn', 'skip']:
-            raise AnsibleLookupError('"on_missing" must be a string and one of "error", "warn" or "skip", not %s' % missing)
-
-        denied = on_denied.lower()
-        if not isinstance(denied, string_types) or denied not in ['error', 'warn', 'skip']:
-            raise AnsibleLookupError('"on_denied" must be a string and one of "error", "warn" or "skip", not %s' % denied)
-
-        credentials = {}
-        if aws_profile:
-            credentials['aws_profile'] = aws_profile
-        else:
-            credentials['aws_profile'] = boto_profile
-        credentials['aws_secret_access_key'] = aws_secret_key
-        credentials['aws_access_key_id'] = aws_access_key
-        credentials['aws_session_token'] = aws_security_token
-
-        # fallback to IAM role credentials
-        if not credentials['aws_profile'] and not (
-                credentials['aws_access_key_id'] and credentials['aws_secret_access_key']):
-            session = botocore.session.get_session()
-            if session.get_credentials() is not None:
-                credentials['aws_access_key_id'] = session.get_credentials().access_key
-                credentials['aws_secret_access_key'] = session.get_credentials().secret_key
-                credentials['aws_session_token'] = session.get_credentials().token
-
-        client = _boto3_conn(region, credentials)
-
-        if bypath:
-            secrets = {}
-            for term in terms:
-                try:
-                    paginator = client.get_paginator('list_secrets')
-                    paginator_response = paginator.paginate(
-                        Filters=[{'Key': 'name', 'Values': [term]}])
-                    for object in paginator_response:
-                        if 'SecretList' in object:
-                            for secret_obj in object['SecretList']:
-                                secrets.update({secret_obj['Name']: self.get_secret_value(
-                                    secret_obj['Name'], client, on_missing=missing, on_denied=denied)})
-                    secrets = [secrets]
-
-                except (botocore.exceptions.ClientError, botocore.exceptions.BotoCoreError) as e:
-                    raise AnsibleLookupError("Failed to retrieve secret: %s" % to_native(e))
-        else:
-            secrets = []
-            for term in terms:
-                value = self.get_secret_value(term, client,
-                                              version_stage=version_stage, version_id=version_id,
-                                              on_missing=missing, on_denied=denied, on_deleted=deleted,
-                                              nested=nested)
-                if value:
-                    secrets.append(value)
-            if join:
-                joined_secret = []
-                joined_secret.append(''.join(secrets))
-                return joined_secret
-
-        return secrets
-
-    def get_secret_value(self, term, client, version_stage=None, version_id=None, on_missing=None, on_denied=None, on_deleted=None, nested=False):
-        params = {}
-        params['SecretId'] = term
-        if version_id:
-            params['VersionId'] = version_id
-        if version_stage:
-            params['VersionStage'] = version_stage
-        if nested:
-            if len(term.split('.')) < 2:
-                raise AnsibleLookupError("Nested query must use the following syntax: `aws_secret_name.<key_name>.<key_name>")
-            secret_name = term.split('.')[0]
-            params['SecretId'] = secret_name
-
-        try:
-            response = client.get_secret_value(**params)
-            if 'SecretBinary' in response:
-                return response['SecretBinary']
-            if 'SecretString' in response:
-                if nested:
-                    query = term.split('.')[1:]
-                    secret_string = json.loads(response['SecretString'])
-                    ret_val = secret_string
-                    for key in query:
-                        if key in ret_val:
-                            ret_val = ret_val[key]
-                        else:
-                            raise AnsibleLookupError("Successfully retrieved secret but there exists no key {0} in the secret".format(key))
-                    return str(ret_val)
-                else:
-                    return response['SecretString']
-        except is_boto3_error_message('marked for deletion'):
-            if on_deleted == 'error':
-                raise AnsibleLookupError("Failed to find secret %s (marked for deletion)" % term)
-            elif on_deleted == 'warn':
-                self._display.warning('Skipping, did not find secret (marked for deletion) %s' % term)
-        except is_boto3_error_code('ResourceNotFoundException'):  # pylint: disable=duplicate-except
-            if on_missing == 'error':
-                raise AnsibleLookupError("Failed to find secret %s (ResourceNotFound)" % term)
-            elif on_missing == 'warn':
-                self._display.warning('Skipping, did not find secret %s' % term)
-        except is_boto3_error_code('AccessDeniedException'):  # pylint: disable=duplicate-except
-            if on_denied == 'error':
-                raise AnsibleLookupError("Failed to access secret %s (AccessDenied)" % term)
-            elif on_denied == 'warn':
-                self._display.warning('Skipping, access denied for secret %s' % term)
-        except (botocore.exceptions.ClientError, botocore.exceptions.BotoCoreError) as e:  # pylint: disable=duplicate-except
-            raise AnsibleLookupError("Failed to retrieve secret: %s" % to_native(e))
-
-        return None
diff --git a/ansible_collections/amazon/aws/plugins/lookup/aws_service_ip_ranges.py b/ansible_collections/amazon/aws/plugins/lookup/aws_service_ip_ranges.py
index 251debf40..c01f583f0 100644
--- a/ansible_collections/amazon/aws/plugins/lookup/aws_service_ip_ranges.py
+++ b/ansible_collections/amazon/aws/plugins/lookup/aws_service_ip_ranges.py
@@ -1,10 +1,10 @@
+# -*- coding: utf-8 -*-
+
 # (c) 2016 James Turner <turnerjsm@gmail.com>
 # (c) 2017 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
 
-DOCUMENTATION = '''
+DOCUMENTATION = r"""
 name: aws_service_ip_ranges
 author:
   - James Turner (!UNKNOWN) <turnerjsm@gmail.com>
@@ -22,23 +22,22 @@ options:
   ipv6_prefixes:
     description: 'When I(ipv6_prefixes=True) the lookup will return ipv6 addresses instead of ipv4 addresses'
     version_added: 2.1.0
-'''
+"""
 
-EXAMPLES = """
+EXAMPLES = r"""
 vars:
   ec2_ranges: "{{ lookup('aws_service_ip_ranges', region='ap-southeast-2', service='EC2', wantlist=True) }}"
 tasks:
+  - name: "use list return option and iterate as a loop"
+    debug: msg="{% for cidr in ec2_ranges %}{{ cidr }} {% endfor %}"
+  # "52.62.0.0/15 52.64.0.0/17 52.64.128.0/17 52.65.0.0/16 52.95.241.0/24 52.95.255.16/28 54.66.0.0/16 "
 
-- name: "use list return option and iterate as a loop"
-  debug: msg="{% for cidr in ec2_ranges %}{{ cidr }} {% endfor %}"
-# "52.62.0.0/15 52.64.0.0/17 52.64.128.0/17 52.65.0.0/16 52.95.241.0/24 52.95.255.16/28 54.66.0.0/16 "
-
-- name: "Pull S3 IP ranges, and print the default return style"
-  debug: msg="{{ lookup('aws_service_ip_ranges', region='us-east-1', service='S3') }}"
-# "52.92.16.0/20,52.216.0.0/15,54.231.0.0/17"
+  - name: "Pull S3 IP ranges, and print the default return style"
+    debug: msg="{{ lookup('aws_service_ip_ranges', region='us-east-1', service='S3') }}"
+  # "52.92.16.0/20,52.216.0.0/15,54.231.0.0/17"
 """
 
-RETURN = """
+RETURN = r"""
 _raw:
   description: comma-separated list of CIDR ranges
 """
@@ -46,12 +45,12 @@ _raw:
 import json
 
 from ansible.errors import AnsibleLookupError
+from ansible.module_utils._text import to_native
 from ansible.module_utils.six.moves.urllib.error import HTTPError
 from ansible.module_utils.six.moves.urllib.error import URLError
-from ansible.module_utils._text import to_native
 from ansible.module_utils.urls import ConnectionError
-from ansible.module_utils.urls import open_url
 from ansible.module_utils.urls import SSLValidationError
+from ansible.module_utils.urls import open_url
 from ansible.plugins.lookup import LookupBase
 
 
@@ -65,26 +64,26 @@ class LookupModule(LookupBase):
             ip_prefix_label = "ip_prefix"
 
         try:
-            resp = open_url('https://ip-ranges.amazonaws.com/ip-ranges.json')
+            resp = open_url("https://ip-ranges.amazonaws.com/ip-ranges.json")
             amazon_response = json.load(resp)[prefixes_label]
-        except getattr(json.decoder, 'JSONDecodeError', ValueError) as e:
+        except getattr(json.decoder, "JSONDecodeError", ValueError) as e:
             # on Python 3+, json.decoder.JSONDecodeError is raised for bad
             # JSON. On 2.x it's a ValueError
-            raise AnsibleLookupError("Could not decode AWS IP ranges: %s" % to_native(e))
+            raise AnsibleLookupError(f"Could not decode AWS IP ranges: {to_native(e)}")
         except HTTPError as e:
-            raise AnsibleLookupError("Received HTTP error while pulling IP ranges: %s" % to_native(e))
+            raise AnsibleLookupError(f"Received HTTP error while pulling IP ranges: {to_native(e)}")
         except SSLValidationError as e:
-            raise AnsibleLookupError("Error validating the server's certificate for: %s" % to_native(e))
+            raise AnsibleLookupError(f"Error validating the server's certificate for: {to_native(e)}")
         except URLError as e:
-            raise AnsibleLookupError("Failed look up IP range service: %s" % to_native(e))
+            raise AnsibleLookupError(f"Failed look up IP range service: {to_native(e)}")
         except ConnectionError as e:
-            raise AnsibleLookupError("Error connecting to IP range service: %s" % to_native(e))
+            raise AnsibleLookupError(f"Error connecting to IP range service: {to_native(e)}")
 
-        if 'region' in kwargs:
-            region = kwargs['region']
-            amazon_response = (item for item in amazon_response if item['region'] == region)
-        if 'service' in kwargs:
-            service = str.upper(kwargs['service'])
-            amazon_response = (item for item in amazon_response if item['service'] == service)
+        if "region" in kwargs:
+            region = kwargs["region"]
+            amazon_response = (item for item in amazon_response if item["region"] == region)
+        if "service" in kwargs:
+            service = str.upper(kwargs["service"])
+            amazon_response = (item for item in amazon_response if item["service"] == service)
         iprange = [item[ip_prefix_label] for item in amazon_response]
         return iprange
diff --git a/ansible_collections/amazon/aws/plugins/lookup/aws_ssm.py b/ansible_collections/amazon/aws/plugins/lookup/aws_ssm.py
deleted file mode 100644
index e71808560..000000000
--- a/ansible_collections/amazon/aws/plugins/lookup/aws_ssm.py
+++ /dev/null
@@ -1,286 +0,0 @@
-# (c) 2016, Bill Wang <ozbillwang(at)gmail.com>
-# (c) 2017, Marat Bakeev <hawara(at)gmail.com>
-# (c) 2018, Michael De La Rue <siblemitcom.mddlr(at)spamgourmet.com>
-# (c) 2017 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
-
-DOCUMENTATION = '''
-name: aws_ssm
-author:
-  - Bill Wang (!UNKNOWN) <ozbillwang(at)gmail.com>
-  - Marat Bakeev (!UNKNOWN) <hawara(at)gmail.com>
-  - Michael De La Rue (!UNKNOWN) <siblemitcom.mddlr@spamgourmet.com>
-short_description: Get the value for a SSM parameter or all parameters under a path
-description:
-  - Get the value for an Amazon Simple Systems Manager parameter or a hierarchy of parameters.
-    The first argument you pass the lookup can either be a parameter name or a hierarchy of
-    parameters. Hierarchies start with a forward slash and end with the parameter name. Up to
-    5 layers may be specified.
-  - If looking up an explicitly listed parameter by name which does not exist then the lookup
-    will generate an error. You can use the ```default``` filter to give a default value in
-    this case but must set the ```on_missing``` parameter to ```skip``` or ```warn```. You must
-    also set the second parameter of the ```default``` filter to ```true``` (see examples below).
-  - When looking up a path for parameters under it a dictionary will be returned for each path.
-    If there is no parameter under that path then the lookup will generate an error.
-  - If the lookup fails due to lack of permissions or due to an AWS client error then the aws_ssm
-    will generate an error. If you want to continue in this case then you will have to set up
-    two ansible tasks, one which sets a variable and ignores failures and one which uses the value
-    of that variable with a default.  See the examples below.
-
-options:
-  decrypt:
-    description: A boolean to indicate whether to decrypt the parameter.
-    default: true
-    type: boolean
-  bypath:
-    description: A boolean to indicate whether the parameter is provided as a hierarchy.
-    default: false
-    type: boolean
-  recursive:
-    description: A boolean to indicate whether to retrieve all parameters within a hierarchy.
-    default: false
-    type: boolean
-  shortnames:
-    description: Indicates whether to return the name only without path if using a parameter hierarchy.
-    default: false
-    type: boolean
-  on_missing:
-    description:
-        - Action to take if the SSM parameter is missing.
-        - C(error) will raise a fatal error when the SSM parameter is missing.
-        - C(skip) will silently ignore the missing SSM parameter.
-        - C(warn) will skip over the missing SSM parameter but issue a warning.
-    default: error
-    type: string
-    choices: ['error', 'skip', 'warn']
-    version_added: 2.0.0
-  on_denied:
-    description:
-        - Action to take if access to the SSM parameter is denied.
-        - C(error) will raise a fatal error when access to the SSM parameter is denied.
-        - C(skip) will silently ignore the denied SSM parameter.
-        - C(warn) will skip over the denied SSM parameter but issue a warning.
-    default: error
-    type: string
-    choices: ['error', 'skip', 'warn']
-    version_added: 2.0.0
-  endpoint:
-    description: Use a custom endpoint when connecting to SSM service.
-    type: string
-    version_added: 3.3.0
-extends_documentation_fragment:
-  - amazon.aws.boto3
-'''
-
-EXAMPLES = '''
-# lookup sample:
-- name: lookup ssm parameter store in the current region
-  debug: msg="{{ lookup('aws_ssm', 'Hello' ) }}"
-
-- name: lookup ssm parameter store in specified region
-  debug: msg="{{ lookup('aws_ssm', 'Hello', region='us-east-2' ) }}"
-
-- name: lookup ssm parameter store without decryption
-  debug: msg="{{ lookup('aws_ssm', 'Hello', decrypt=False ) }}"
-
-- name: lookup ssm parameter store using a specified aws profile
-  debug: msg="{{ lookup('aws_ssm', 'Hello', aws_profile='myprofile' ) }}"
-
-- name: lookup ssm parameter store using explicit aws credentials
-  debug: msg="{{ lookup('aws_ssm', 'Hello', aws_access_key=my_aws_access_key, aws_secret_key=my_aws_secret_key, aws_security_token=my_security_token ) }}"
-
-- name: lookup ssm parameter store with all options
-  debug: msg="{{ lookup('aws_ssm', 'Hello', decrypt=false, region='us-east-2', aws_profile='myprofile') }}"
-
-- name: lookup ssm parameter and fail if missing
-  debug: msg="{{ lookup('aws_ssm', 'missing-parameter') }}"
-
-- name: lookup a key which doesn't exist, returning a default ('root')
-  debug: msg="{{ lookup('aws_ssm', 'AdminID', on_missing="skip") | default('root', true) }}"
-
-- name: lookup a key which doesn't exist failing to store it in a fact
-  set_fact:
-    temp_secret: "{{ lookup('aws_ssm', '/NoAccess/hiddensecret') }}"
-  ignore_errors: true
-
-- name: show fact default to "access failed" if we don't have access
-  debug: msg="{{ 'the secret was:' ~ temp_secret | default('could not access secret') }}"
-
-- name: return a dictionary of ssm parameters from a hierarchy path
-  debug: msg="{{ lookup('aws_ssm', '/PATH/to/params', region='ap-southeast-2', bypath=true, recursive=true ) }}"
-
-- name: return a dictionary of ssm parameters from a hierarchy path with shortened names (param instead of /PATH/to/param)
-  debug: msg="{{ lookup('aws_ssm', '/PATH/to/params', region='ap-southeast-2', shortnames=true, bypath=true, recursive=true ) }}"
-
-- name: Iterate over a parameter hierarchy (one iteration per parameter)
-  debug: msg='Key contains {{ item.key }} , with value {{ item.value }}'
-  loop: '{{ lookup("aws_ssm", "/demo/", region="ap-southeast-2", bypath=True) | dict2items }}'
-
-- name: Iterate over multiple paths as dictionaries (one iteration per path)
-  debug: msg='Path contains {{ item }}'
-  loop: '{{ lookup("aws_ssm", "/demo/", "/demo1/", bypath=True)}}'
-
-- name: lookup ssm parameter warn if access is denied
-  debug: msg="{{ lookup('aws_ssm', 'missing-parameter', on_denied="warn" ) }}"
-'''
-
-try:
-    import botocore
-except ImportError:
-    pass  # will be captured by imported HAS_BOTO3
-
-from ansible.errors import AnsibleLookupError
-from ansible.module_utils._text import to_native
-from ansible.plugins.lookup import LookupBase
-from ansible.utils.display import Display
-from ansible.module_utils.six import string_types
-from ansible.module_utils.basic import missing_required_lib
-
-from ansible_collections.amazon.aws.plugins.module_utils.ec2 import boto3_conn
-from ansible_collections.amazon.aws.plugins.module_utils.ec2 import get_aws_connection_info
-from ansible_collections.amazon.aws.plugins.module_utils.ec2 import HAS_BOTO3
-from ansible_collections.amazon.aws.plugins.module_utils.ec2 import boto3_tag_list_to_ansible_dict
-from ansible_collections.amazon.aws.plugins.module_utils.core import is_boto3_error_code
-
-display = Display()
-
-
-class LookupModule(LookupBase):
-    def run(self, terms, variables=None, boto_profile=None, aws_profile=None,
-            aws_secret_key=None, aws_access_key=None, aws_security_token=None, region=None,
-            bypath=False, shortnames=False, recursive=False, decrypt=True, on_missing="error",
-            on_denied="error", endpoint=None):
-        '''
-            :arg terms: a list of lookups to run.
-                e.g. ['parameter_name', 'parameter_name_too' ]
-            :kwarg variables: ansible variables active at the time of the lookup
-            :kwarg aws_secret_key: identity of the AWS key to use
-            :kwarg aws_access_key: AWS secret key (matching identity)
-            :kwarg aws_security_token: AWS session key if using STS
-            :kwarg decrypt: Set to True to get decrypted parameters
-            :kwarg region: AWS region in which to do the lookup
-            :kwarg bypath: Set to True to do a lookup of variables under a path
-            :kwarg recursive: Set to True to recurse below the path (requires bypath=True)
-            :kwarg on_missing: Action to take if the SSM parameter is missing
-            :kwarg on_denied: Action to take if access to the SSM parameter is denied
-            :kwarg endpoint: Endpoint for SSM client
-            :returns: A list of parameter values or a list of dictionaries if bypath=True.
-        '''
-
-        if not HAS_BOTO3:
-            raise AnsibleLookupError(missing_required_lib('botocore and boto3'))
-
-        # validate arguments 'on_missing' and 'on_denied'
-        if on_missing is not None and (not isinstance(on_missing, string_types) or on_missing.lower() not in ['error', 'warn', 'skip']):
-            raise AnsibleLookupError('"on_missing" must be a string and one of "error", "warn" or "skip", not %s' % on_missing)
-        if on_denied is not None and (not isinstance(on_denied, string_types) or on_denied.lower() not in ['error', 'warn', 'skip']):
-            raise AnsibleLookupError('"on_denied" must be a string and one of "error", "warn" or "skip", not %s' % on_denied)
-
-        ret = []
-        ssm_dict = {}
-
-        self.params = variables
-
-        cli_region, cli_endpoint, cli_boto_params = get_aws_connection_info(self, boto3=True)
-
-        if region:
-            cli_region = region
-
-        if endpoint:
-            cli_endpoint = endpoint
-
-        # For backward  compatibility
-        if aws_access_key:
-            cli_boto_params.update({'aws_access_key_id': aws_access_key})
-        if aws_secret_key:
-            cli_boto_params.update({'aws_secret_access_key': aws_secret_key})
-        if aws_security_token:
-            cli_boto_params.update({'aws_session_token': aws_security_token})
-        if boto_profile:
-            cli_boto_params.update({'profile_name': boto_profile})
-        if aws_profile:
-            cli_boto_params.update({'profile_name': aws_profile})
-
-        cli_boto_params.update(dict(
-            conn_type='client',
-            resource='ssm',
-            region=cli_region,
-            endpoint=cli_endpoint,
-        ))
-
-        client = boto3_conn(module=self, **cli_boto_params)
-
-        ssm_dict['WithDecryption'] = decrypt
-
-        # Lookup by path
-        if bypath:
-            ssm_dict['Recursive'] = recursive
-            for term in terms:
-                display.vvv("AWS_ssm path lookup term: %s in region: %s" % (term, region))
-
-                paramlist = self.get_path_parameters(client, ssm_dict, term, on_missing.lower(), on_denied.lower())
-                # Shorten parameter names. Yes, this will return
-                # duplicate names with different values.
-                if shortnames:
-                    for x in paramlist:
-                        x['Name'] = x['Name'][x['Name'].rfind('/') + 1:]
-
-                display.vvvv("AWS_ssm path lookup returned: %s" % str(paramlist))
-
-                ret.append(boto3_tag_list_to_ansible_dict(paramlist,
-                                                          tag_name_key_name="Name",
-                                                          tag_value_key_name="Value"))
-        # Lookup by parameter name - always returns a list with one or
-        # no entry.
-        else:
-            display.vvv("AWS_ssm name lookup term: %s" % terms)
-            for term in terms:
-                ret.append(self.get_parameter_value(client, ssm_dict, term, on_missing.lower(), on_denied.lower()))
-        display.vvvv("AWS_ssm path lookup returning: %s " % str(ret))
-        return ret
-
-    def get_path_parameters(self, client, ssm_dict, term, on_missing, on_denied):
-        ssm_dict["Path"] = term
-        paginator = client.get_paginator('get_parameters_by_path')
-        try:
-            paramlist = paginator.paginate(**ssm_dict).build_full_result()['Parameters']
-        except is_boto3_error_code('AccessDeniedException'):
-            if on_denied == 'error':
-                raise AnsibleLookupError("Failed to access SSM parameter path %s (AccessDenied)" % term)
-            elif on_denied == 'warn':
-                self._display.warning('Skipping, access denied for SSM parameter path %s' % term)
-                paramlist = [{}]
-            elif on_denied == 'skip':
-                paramlist = [{}]
-        except botocore.exceptions.ClientError as e:  # pylint: disable=duplicate-except
-            raise AnsibleLookupError("SSM lookup exception: {0}".format(to_native(e)))
-
-        if not len(paramlist):
-            if on_missing == "error":
-                raise AnsibleLookupError("Failed to find SSM parameter path %s (ResourceNotFound)" % term)
-            elif on_missing == "warn":
-                self._display.warning('Skipping, did not find SSM parameter path %s' % term)
-
-        return paramlist
-
-    def get_parameter_value(self, client, ssm_dict, term, on_missing, on_denied):
-        ssm_dict["Name"] = term
-        try:
-            response = client.get_parameter(**ssm_dict)
-            return response['Parameter']['Value']
-        except is_boto3_error_code('ParameterNotFound'):
-            if on_missing == 'error':
-                raise AnsibleLookupError("Failed to find SSM parameter %s (ResourceNotFound)" % term)
-            elif on_missing == 'warn':
-                self._display.warning('Skipping, did not find SSM parameter %s' % term)
-        except is_boto3_error_code('AccessDeniedException'):  # pylint: disable=duplicate-except
-            if on_denied == 'error':
-                raise AnsibleLookupError("Failed to access SSM parameter %s (AccessDenied)" % term)
-            elif on_denied == 'warn':
-                self._display.warning('Skipping, access denied for SSM parameter %s' % term)
-        except botocore.exceptions.ClientError as e:  # pylint: disable=duplicate-except
-            raise AnsibleLookupError("SSM lookup exception: {0}".format(to_native(e)))
-        return None
diff --git a/ansible_collections/amazon/aws/plugins/lookup/secretsmanager_secret.py b/ansible_collections/amazon/aws/plugins/lookup/secretsmanager_secret.py
new file mode 100644
index 000000000..06ad10be5
--- /dev/null
+++ b/ansible_collections/amazon/aws/plugins/lookup/secretsmanager_secret.py
@@ -0,0 +1,294 @@
+# -*- coding: utf-8 -*-
+
+# Copyright: (c) 2018, Aaron Smith <ajsmith10381@gmail.com>
+# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)
+
+DOCUMENTATION = r"""
+name: secretsmanager_secret
+author:
+  - Aaron Smith (!UNKNOWN) <ajsmith10381@gmail.com>
+
+short_description: Look up secrets stored in AWS Secrets Manager
+description:
+  - Look up secrets stored in AWS Secrets Manager provided the caller
+    has the appropriate permissions to read the secret.
+  - Lookup is based on the secret's I(Name) value.
+  - Optional parameters can be passed into this lookup; I(version_id) and I(version_stage)
+  - Prior to release 6.0.0 this module was known as C(aws_ssm), the usage remains the same.
+
+options:
+  _terms:
+    description: Name of the secret to look up in AWS Secrets Manager.
+    required: True
+  bypath:
+    description: A boolean to indicate whether the parameter is provided as a hierarchy.
+    default: false
+    type: boolean
+    version_added: 1.4.0
+  nested:
+    description: A boolean to indicate the secret contains nested values.
+    type: boolean
+    default: false
+    version_added: 1.4.0
+  version_id:
+    description: Version of the secret(s).
+    required: False
+  version_stage:
+    description: Stage of the secret version.
+    required: False
+  join:
+    description:
+      - Join two or more entries to form an extended secret.
+      - This is useful for overcoming the 4096 character limit imposed by AWS.
+      - No effect when used with I(bypath).
+    type: boolean
+    default: false
+  on_deleted:
+    description:
+      - Action to take if the secret has been marked for deletion.
+      - C(error) will raise a fatal error when the secret has been marked for deletion.
+      - C(skip) will silently ignore the deleted secret.
+      - C(warn) will skip over the deleted secret but issue a warning.
+    default: error
+    type: string
+    choices: ['error', 'skip', 'warn']
+    version_added: 2.0.0
+  on_missing:
+    description:
+      - Action to take if the secret is missing.
+      - C(error) will raise a fatal error when the secret is missing.
+      - C(skip) will silently ignore the missing secret.
+      - C(warn) will skip over the missing secret but issue a warning.
+    default: error
+    type: string
+    choices: ['error', 'skip', 'warn']
+  on_denied:
+    description:
+      - Action to take if access to the secret is denied.
+      - C(error) will raise a fatal error when access to the secret is denied.
+      - C(skip) will silently ignore the denied secret.
+      - C(warn) will skip over the denied secret but issue a warning.
+    default: error
+    type: string
+    choices: ['error', 'skip', 'warn']
+extends_documentation_fragment:
+  - amazon.aws.boto3
+  - amazon.aws.common.plugins
+  - amazon.aws.region.plugins
+"""
+
+EXAMPLES = r"""
+- name: lookup secretsmanager secret in the current region
+  debug: msg="{{ lookup('amazon.aws.aws_secret', '/path/to/secrets', bypath=true) }}"
+
+- name: Create RDS instance with aws_secret lookup for password param
+  rds:
+    command: create
+    instance_name: app-db
+    db_engine: MySQL
+    size: 10
+    instance_type: db.m1.small
+    username: dbadmin
+    password: "{{ lookup('amazon.aws.aws_secret', 'DbSecret') }}"
+    tags:
+      Environment: staging
+
+- name: skip if secret does not exist
+  debug: msg="{{ lookup('amazon.aws.aws_secret', 'secret-not-exist', on_missing='skip')}}"
+
+- name: warn if access to the secret is denied
+  debug: msg="{{ lookup('amazon.aws.aws_secret', 'secret-denied', on_denied='warn')}}"
+
+- name: lookup secretsmanager secret in the current region using the nested feature
+  debug: msg="{{ lookup('amazon.aws.aws_secret', 'secrets.environments.production.password', nested=true) }}"
+  # The secret can be queried using the following syntax: `aws_secret_object_name.key1.key2.key3`.
+  # If an object is of the form `{"key1":{"key2":{"key3":1}}}` the query would return the value `1`.
+- name: lookup secretsmanager secret in a specific region using specified region and aws profile using nested feature
+  debug: >
+   msg="{{ lookup('amazon.aws.aws_secret', 'secrets.environments.production.password', region=region, profile=aws_profile,
+   access_key=aws_access_key, secret_key=aws_secret_key, nested=true) }}"
+  # The secret can be queried using the following syntax: `aws_secret_object_name.key1.key2.key3`.
+  # If an object is of the form `{"key1":{"key2":{"key3":1}}}` the query would return the value `1`.
+  # Region is the AWS region where the AWS secret is stored.
+  # AWS_profile is the aws profile to use, that has access to the AWS secret.
+"""
+
+RETURN = r"""
+_raw:
+  description:
+    Returns the value of the secret stored in AWS Secrets Manager.
+"""
+
+import json
+
+try:
+    import botocore
+except ImportError:
+    pass  # Handled by AWSLookupBase
+
+from ansible.errors import AnsibleLookupError
+from ansible.module_utils._text import to_native
+from ansible.module_utils.six import string_types
+
+from ansible_collections.amazon.aws.plugins.module_utils.botocore import is_boto3_error_code
+from ansible_collections.amazon.aws.plugins.module_utils.botocore import is_boto3_error_message
+from ansible_collections.amazon.aws.plugins.module_utils.retries import AWSRetry
+from ansible_collections.amazon.aws.plugins.plugin_utils.lookup import AWSLookupBase
+
+
+def _list_secrets(client, term):
+    paginator = client.get_paginator("list_secrets")
+    return paginator.paginate(Filters=[{"Key": "name", "Values": [term]}])
+
+
+class LookupModule(AWSLookupBase):
+    def run(self, terms, variables, **kwargs):
+        """
+        :arg terms: a list of lookups to run.
+            e.g. ['example_secret_name', 'example_secret_too' ]
+        :variables: ansible variables active at the time of the lookup
+        :returns: A list of parameter values or a list of dictionaries if bypath=True.
+        """
+
+        super().run(terms, variables, **kwargs)
+
+        on_missing = self.get_option("on_missing")
+        on_denied = self.get_option("on_denied")
+        on_deleted = self.get_option("on_deleted")
+
+        # validate arguments 'on_missing' and 'on_denied'
+        if on_missing is not None and (
+            not isinstance(on_missing, string_types) or on_missing.lower() not in ["error", "warn", "skip"]
+        ):
+            raise AnsibleLookupError(
+                f'"on_missing" must be a string and one of "error", "warn" or "skip", not {on_missing}'
+            )
+        if on_denied is not None and (
+            not isinstance(on_denied, string_types) or on_denied.lower() not in ["error", "warn", "skip"]
+        ):
+            raise AnsibleLookupError(
+                f'"on_denied" must be a string and one of "error", "warn" or "skip", not {on_denied}'
+            )
+        if on_deleted is not None and (
+            not isinstance(on_deleted, string_types) or on_deleted.lower() not in ["error", "warn", "skip"]
+        ):
+            raise AnsibleLookupError(
+                f'"on_deleted" must be a string and one of "error", "warn" or "skip", not {on_deleted}'
+            )
+
+        client = self.client("secretsmanager", AWSRetry.jittered_backoff())
+
+        if self.get_option("bypath"):
+            secrets = {}
+            for term in terms:
+                try:
+                    for object in _list_secrets(client, term):
+                        if "SecretList" in object:
+                            for secret_obj in object["SecretList"]:
+                                secrets.update(
+                                    {
+                                        secret_obj["Name"]: self.get_secret_value(
+                                            secret_obj["Name"], client, on_missing=on_missing, on_denied=on_denied
+                                        )
+                                    }
+                                )
+                    secrets = [secrets]
+
+                except (botocore.exceptions.ClientError, botocore.exceptions.BotoCoreError) as e:
+                    raise AnsibleLookupError(f"Failed to retrieve secret: {to_native(e)}")
+        else:
+            secrets = []
+            for term in terms:
+                value = self.get_secret_value(
+                    term,
+                    client,
+                    version_stage=self.get_option("version_stage"),
+                    version_id=self.get_option("version_id"),
+                    on_missing=on_missing,
+                    on_denied=on_denied,
+                    on_deleted=on_deleted,
+                    nested=self.get_option("nested"),
+                )
+                if value:
+                    secrets.append(value)
+            if self.get_option("join"):
+                joined_secret = []
+                joined_secret.append("".join(secrets))
+                return joined_secret
+
+        return secrets
+
+    def get_secret_value(
+        self,
+        term,
+        client,
+        version_stage=None,
+        version_id=None,
+        on_missing=None,
+        on_denied=None,
+        on_deleted=None,
+        nested=False,
+    ):
+        params = {}
+        params["SecretId"] = term
+        if version_id:
+            params["VersionId"] = version_id
+        if version_stage:
+            params["VersionStage"] = version_stage
+        if nested:
+            if len(term.split(".")) < 2:
+                raise AnsibleLookupError(
+                    "Nested query must use the following syntax: `aws_secret_name.<key_name>.<key_name>"
+                )
+            secret_name = term.split(".")[0]
+            params["SecretId"] = secret_name
+
+        try:
+            response = client.get_secret_value(aws_retry=True, **params)
+            if "SecretBinary" in response:
+                return response["SecretBinary"]
+            if "SecretString" in response:
+                if nested:
+                    query = term.split(".")[1:]
+                    path = None
+                    secret_string = json.loads(response["SecretString"])
+                    ret_val = secret_string
+                    while query:
+                        key = query.pop(0)
+                        path = key if not path else path + "." + key
+                        if key in ret_val:
+                            ret_val = ret_val[key]
+                        elif on_missing == "warn":
+                            self._display.warning(
+                                f"Skipping, Successfully retrieved secret but there exists no key {path} in the secret"
+                            )
+                            return None
+                        elif on_missing == "error":
+                            raise AnsibleLookupError(
+                                f"Successfully retrieved secret but there exists no key {path} in the secret"
+                            )
+                    return str(ret_val)
+                else:
+                    return response["SecretString"]
+        except is_boto3_error_message("marked for deletion"):
+            if on_deleted == "error":
+                raise AnsibleLookupError(f"Failed to find secret {term} (marked for deletion)")
+            elif on_deleted == "warn":
+                self._display.warning(f"Skipping, did not find secret (marked for deletion) {term}")
+        except is_boto3_error_code("ResourceNotFoundException"):  # pylint: disable=duplicate-except
+            if on_missing == "error":
+                raise AnsibleLookupError(f"Failed to find secret {term} (ResourceNotFound)")
+            elif on_missing == "warn":
+                self._display.warning(f"Skipping, did not find secret {term}")
+        except is_boto3_error_code("AccessDeniedException"):  # pylint: disable=duplicate-except
+            if on_denied == "error":
+                raise AnsibleLookupError(f"Failed to access secret {term} (AccessDenied)")
+            elif on_denied == "warn":
+                self._display.warning(f"Skipping, access denied for secret {term}")
+        except (
+            botocore.exceptions.ClientError,
+            botocore.exceptions.BotoCoreError,
+        ) as e:  # pylint: disable=duplicate-except
+            raise AnsibleLookupError(f"Failed to retrieve secret: {to_native(e)}")
+
+        return None
diff --git a/ansible_collections/amazon/aws/plugins/lookup/ssm_parameter.py b/ansible_collections/amazon/aws/plugins/lookup/ssm_parameter.py
new file mode 100644
index 000000000..0ca3afdd8
--- /dev/null
+++ b/ansible_collections/amazon/aws/plugins/lookup/ssm_parameter.py
@@ -0,0 +1,251 @@
+# -*- coding: utf-8 -*-
+
+# (c) 2016, Bill Wang <ozbillwang(at)gmail.com>
+# (c) 2017, Marat Bakeev <hawara(at)gmail.com>
+# (c) 2018, Michael De La Rue <siblemitcom.mddlr(at)spamgourmet.com>
+# (c) 2017 Ansible Project
+# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)
+
+DOCUMENTATION = r"""
+name: ssm_parameter
+author:
+  - Bill Wang (!UNKNOWN) <ozbillwang(at)gmail.com>
+  - Marat Bakeev (!UNKNOWN) <hawara(at)gmail.com>
+  - Michael De La Rue (!UNKNOWN) <siblemitcom.mddlr@spamgourmet.com>
+short_description: gets the value for a SSM parameter or all parameters under a path
+description:
+  - Get the value for an Amazon Simple Systems Manager parameter or a hierarchy of parameters.
+    The first argument you pass the lookup can either be a parameter name or a hierarchy of
+    parameters. Hierarchies start with a forward slash and end with the parameter name. Up to
+    5 layers may be specified.
+  - If looking up an explicitly listed parameter by name which does not exist then the lookup
+    will generate an error. You can use the C(default) filter to give a default value in
+    this case but must set the I(on_missing) parameter to C(skip) or C(warn). You must
+    also set the second parameter of the C(default) filter to C(true) (see examples below).
+  - When looking up a path for parameters under it a dictionary will be returned for each path.
+    If there is no parameter under that path then the lookup will generate an error.
+  - If the lookup fails due to lack of permissions or due to an AWS client error then the aws_ssm
+    will generate an error. If you want to continue in this case then you will have to set up
+    two ansible tasks, one which sets a variable and ignores failures and one which uses the value
+    of that variable with a default.  See the examples below.
+  - Prior to release 6.0.0 this module was known as C(aws_ssm), the usage remains the same.
+
+options:
+  decrypt:
+    description: A boolean to indicate whether to decrypt the parameter.
+    default: true
+    type: boolean
+  bypath:
+    description: A boolean to indicate whether the parameter is provided as a hierarchy.
+    default: false
+    type: boolean
+  recursive:
+    description: A boolean to indicate whether to retrieve all parameters within a hierarchy.
+    default: false
+    type: boolean
+  shortnames:
+    description: Indicates whether to return the name only without path if using a parameter hierarchy.
+    default: false
+    type: boolean
+  on_missing:
+    description:
+        - Action to take if the SSM parameter is missing.
+        - C(error) will raise a fatal error when the SSM parameter is missing.
+        - C(skip) will silently ignore the missing SSM parameter.
+        - C(warn) will skip over the missing SSM parameter but issue a warning.
+    default: error
+    type: string
+    choices: ['error', 'skip', 'warn']
+    version_added: 2.0.0
+  on_denied:
+    description:
+        - Action to take if access to the SSM parameter is denied.
+        - C(error) will raise a fatal error when access to the SSM parameter is denied.
+        - C(skip) will silently ignore the denied SSM parameter.
+        - C(warn) will skip over the denied SSM parameter but issue a warning.
+    default: error
+    type: string
+    choices: ['error', 'skip', 'warn']
+    version_added: 2.0.0
+extends_documentation_fragment:
+  - amazon.aws.boto3
+  - amazon.aws.common.plugins
+  - amazon.aws.region.plugins
+"""
+
+EXAMPLES = r"""
+# lookup sample:
+- name: lookup ssm parameter store in the current region
+  debug: msg="{{ lookup('amazon.aws.aws_ssm', 'Hello' ) }}"
+
+- name: lookup ssm parameter store in specified region
+  debug: msg="{{ lookup('amazon.aws.aws_ssm', 'Hello', region='us-east-2' ) }}"
+
+- name: lookup ssm parameter store without decryption
+  debug: msg="{{ lookup('amazon.aws.aws_ssm', 'Hello', decrypt=False ) }}"
+
+- name: lookup ssm parameter store using a specified aws profile
+  debug: msg="{{ lookup('amazon.aws.aws_ssm', 'Hello', profile='myprofile' ) }}"
+
+- name: lookup ssm parameter store using explicit aws credentials
+  debug:
+    msg: >-
+      {{ lookup('amazon.aws.aws_ssm', 'Hello', access_key=my_aws_access_key, secret_key=my_aws_secret_key, session_token=my_session_token ) }}"
+
+- name: lookup ssm parameter store with all options
+  debug: msg="{{ lookup('amazon.aws.aws_ssm', 'Hello', decrypt=false, region='us-east-2', profile='myprofile') }}"
+
+- name: lookup ssm parameter and fail if missing
+  debug: msg="{{ lookup('amazon.aws.aws_ssm', 'missing-parameter') }}"
+
+- name: lookup a key which doesn't exist, returning a default ('root')
+  debug: msg="{{ lookup('amazon.aws.aws_ssm', 'AdminID', on_missing="skip") | default('root', true) }}"
+
+- name: lookup a key which doesn't exist failing to store it in a fact
+  set_fact:
+    temp_secret: "{{ lookup('amazon.aws.aws_ssm', '/NoAccess/hiddensecret') }}"
+  ignore_errors: true
+
+- name: show fact default to "access failed" if we don't have access
+  debug: msg="{{ 'the secret was:' ~ temp_secret | default('could not access secret') }}"
+
+- name: return a dictionary of ssm parameters from a hierarchy path
+  debug: msg="{{ lookup('amazon.aws.aws_ssm', '/PATH/to/params', region='ap-southeast-2', bypath=true, recursive=true ) }}"
+
+- name: return a dictionary of ssm parameters from a hierarchy path with shortened names (param instead of /PATH/to/param)
+  debug: msg="{{ lookup('amazon.aws.aws_ssm', '/PATH/to/params', region='ap-southeast-2', shortnames=true, bypath=true, recursive=true ) }}"
+
+- name: Iterate over a parameter hierarchy (one iteration per parameter)
+  debug: msg='Key contains {{ item.key }} , with value {{ item.value }}'
+  loop: "{{ lookup('amazon.aws.aws_ssm', '/demo/', region='ap-southeast-2', bypath=True) | dict2items }}"
+
+- name: Iterate over multiple paths as dictionaries (one iteration per path)
+  debug: msg='Path contains {{ item }}'
+  loop: "{{ lookup('amazon.aws.aws_ssm', '/demo/', '/demo1/', bypath=True)}}"
+
+- name: lookup ssm parameter warn if access is denied
+  debug: msg="{{ lookup('amazon.aws.aws_ssm', 'missing-parameter', on_denied="warn" ) }}"
+"""
+
+try:
+    import botocore
+except ImportError:
+    pass  # Handled by AWSLookupBase
+
+from ansible.errors import AnsibleLookupError
+from ansible.module_utils._text import to_native
+from ansible.module_utils.six import string_types
+from ansible.utils.display import Display
+
+from ansible_collections.amazon.aws.plugins.module_utils.botocore import is_boto3_error_code
+from ansible_collections.amazon.aws.plugins.module_utils.retries import AWSRetry
+from ansible_collections.amazon.aws.plugins.module_utils.tagging import boto3_tag_list_to_ansible_dict
+from ansible_collections.amazon.aws.plugins.plugin_utils.lookup import AWSLookupBase
+
+display = Display()
+
+
+class LookupModule(AWSLookupBase):
+    def run(self, terms, variables, **kwargs):
+        """
+        :arg terms: a list of lookups to run.
+            e.g. ['parameter_name', 'parameter_name_too' ]
+        :kwarg variables: ansible variables active at the time of the lookup
+        :returns: A list of parameter values or a list of dictionaries if bypath=True.
+        """
+
+        super().run(terms, variables, **kwargs)
+
+        on_missing = self.get_option("on_missing")
+        on_denied = self.get_option("on_denied")
+
+        # validate arguments 'on_missing' and 'on_denied'
+        if on_missing is not None and (
+            not isinstance(on_missing, string_types) or on_missing.lower() not in ["error", "warn", "skip"]
+        ):
+            raise AnsibleLookupError(
+                f'"on_missing" must be a string and one of "error", "warn" or "skip", not {on_missing}'
+            )
+        if on_denied is not None and (
+            not isinstance(on_denied, string_types) or on_denied.lower() not in ["error", "warn", "skip"]
+        ):
+            raise AnsibleLookupError(
+                f'"on_denied" must be a string and one of "error", "warn" or "skip", not {on_denied}'
+            )
+
+        ret = []
+        ssm_dict = {}
+
+        client = self.client("ssm", AWSRetry.jittered_backoff())
+
+        ssm_dict["WithDecryption"] = self.get_option("decrypt")
+
+        # Lookup by path
+        if self.get_option("bypath"):
+            ssm_dict["Recursive"] = self.get_option("recursive")
+            for term in terms:
+                display.vvv(f"AWS_ssm path lookup term: {term} in region: {self.region}")
+
+                paramlist = self.get_path_parameters(client, ssm_dict, term, on_missing.lower(), on_denied.lower())
+                # Shorten parameter names. Yes, this will return
+                # duplicate names with different values.
+                if self.get_option("shortnames"):
+                    for x in paramlist:
+                        x["Name"] = x["Name"][x["Name"].rfind("/") + 1:]  # fmt: skip
+
+                display.vvvv(f"AWS_ssm path lookup returned: {to_native(paramlist)}")
+
+                ret.append(
+                    boto3_tag_list_to_ansible_dict(paramlist, tag_name_key_name="Name", tag_value_key_name="Value")
+                )
+        # Lookup by parameter name - always returns a list with one or
+        # no entry.
+        else:
+            display.vvv(f"AWS_ssm name lookup term: {terms}")
+            for term in terms:
+                ret.append(self.get_parameter_value(client, ssm_dict, term, on_missing.lower(), on_denied.lower()))
+        display.vvvv(f"AWS_ssm path lookup returning: {to_native(ret)} ")
+        return ret
+
+    def get_path_parameters(self, client, ssm_dict, term, on_missing, on_denied):
+        ssm_dict["Path"] = term
+        paginator = client.get_paginator("get_parameters_by_path")
+        try:
+            paramlist = paginator.paginate(**ssm_dict).build_full_result()["Parameters"]
+        except is_boto3_error_code("AccessDeniedException"):
+            if on_denied == "error":
+                raise AnsibleLookupError(f"Failed to access SSM parameter path {term} (AccessDenied)")
+            elif on_denied == "warn":
+                self.warn(f"Skipping, access denied for SSM parameter path {term}")
+                paramlist = [{}]
+            elif on_denied == "skip":
+                paramlist = [{}]
+        except botocore.exceptions.ClientError as e:  # pylint: disable=duplicate-except
+            raise AnsibleLookupError(f"SSM lookup exception: {to_native(e)}")
+
+        if not len(paramlist):
+            if on_missing == "error":
+                raise AnsibleLookupError(f"Failed to find SSM parameter path {term} (ResourceNotFound)")
+            elif on_missing == "warn":
+                self.warn(f"Skipping, did not find SSM parameter path {term}")
+
+        return paramlist
+
+    def get_parameter_value(self, client, ssm_dict, term, on_missing, on_denied):
+        ssm_dict["Name"] = term
+        try:
+            response = client.get_parameter(aws_retry=True, **ssm_dict)
+            return response["Parameter"]["Value"]
+        except is_boto3_error_code("ParameterNotFound"):
+            if on_missing == "error":
+                raise AnsibleLookupError(f"Failed to find SSM parameter {term} (ResourceNotFound)")
+            elif on_missing == "warn":
+                self.warn(f"Skipping, did not find SSM parameter {term}")
+        except is_boto3_error_code("AccessDeniedException"):  # pylint: disable=duplicate-except
+            if on_denied == "error":
+                raise AnsibleLookupError(f"Failed to access SSM parameter {term} (AccessDenied)")
+            elif on_denied == "warn":
+                self.warn(f"Skipping, access denied for SSM parameter {term}")
+        except botocore.exceptions.ClientError as e:  # pylint: disable=duplicate-except
+            raise AnsibleLookupError(f"SSM lookup exception: {to_native(e)}")
+        return None