diff --git a/changelogs/fragments/1347-s3-object-lock-enabled.yml b/changelogs/fragments/1347-s3-object-lock-enabled.yml
new file mode 100644
index 00000000000..bfbd50ce662
--- /dev/null
+++ b/changelogs/fragments/1347-s3-object-lock-enabled.yml
@@ -0,0 +1,2 @@
+minor_changes:
+- s3_bucket - add option to support creation of buckets with object lock enabled (https://github.com/ansible-collections/amazon.aws/pull/1372).
diff --git a/plugins/modules/s3_bucket.py b/plugins/modules/s3_bucket.py
index aec60e3ebe8..3c084d1042e 100644
--- a/plugins/modules/s3_bucket.py
+++ b/plugins/modules/s3_bucket.py
@@ -136,6 +136,12 @@
     choices: [ 'BucketOwnerEnforced', 'BucketOwnerPreferred', 'ObjectWriter' ]
     type: str
     version_added: 2.0.0
+  object_lock_enabled:
+    description:
+      - Whether S3 Object Lock to be enabled.
+      - Defaults to C(False) when creating a new bucket.
+    type: bool
+    version_added: 5.3.0
   delete_object_ownership:
     description:
       - Delete bucket's ownership controls.
@@ -383,6 +389,7 @@ def create_or_update_bucket(s3_client, module, location):
     delete_public_access = module.params.get("delete_public_access")
     delete_object_ownership = module.params.get("delete_object_ownership")
     object_ownership = module.params.get("object_ownership")
+    object_lock_enabled = module.params.get("object_lock_enabled")
     acl = module.params.get("acl")
     changed = False
     result = {}
@@ -396,7 +403,7 @@ def create_or_update_bucket(s3_client, module, location):
 
     if not bucket_is_present:
         try:
-            bucket_changed = create_bucket(s3_client, name, location)
+            bucket_changed = create_bucket(s3_client, name, location, object_lock_enabled)
             s3_client.get_waiter('bucket_exists').wait(Bucket=name)
             changed = changed or bucket_changed
         except botocore.exceptions.WaiterError as e:
@@ -644,6 +651,32 @@ def create_or_update_bucket(s3_client, module, location):
         except (botocore.exceptions.BotoCoreError, botocore.exceptions.ClientError) as e:  # pylint: disable=duplicate-except
             module.fail_json_aws(e, msg="Failed to update bucket ACL")
 
+    # -- Object Lock
+    try:
+        object_lock_status = get_bucket_object_lock_enabled(s3_client, name)
+        result["object_lock_enabled"] = object_lock_status
+    except is_boto3_error_code(["NotImplemented", "XNotImplemented"]) as e:
+        if object_lock_enabled is not None:
+            module.fail_json(msg="Fetching bucket object lock state is not supported")
+    except is_boto3_error_code("ObjectLockConfigurationNotFoundError"):  # pylint: disable=duplicate-except
+        if object_lock_enabled:
+            module.fail_json(msg="Enabling object lock for existing buckets is not supported")
+        result["object_lock_enabled"] = False
+    except is_boto3_error_code("AccessDenied") as e:  # pylint: disable=duplicate-except
+        if object_lock_enabled is not None:
+            module.fail_json(msg="Permission denied fetching object lock state for bucket")
+    except (
+        botocore.exceptions.BotoCoreError,
+        botocore.exceptions.ClientError,
+    ) as e:  # pylint: disable=duplicate-except
+        module.fail_json_aws(e, msg="Failed to fetch bucket object lock state")
+    else:
+        if object_lock_status is not None:
+            if not object_lock_enabled and object_lock_status:
+                module.fail_json(msg="Disabling object lock for existing buckets is not supported")
+            if object_lock_enabled and not object_lock_status:
+                module.fail_json(msg="Enabling object lock for existing buckets is not supported")
+
     # Module exit
     module.exit_json(changed=changed, name=name, **result)
 
@@ -658,15 +691,22 @@ def bucket_exists(s3_client, bucket_name):
 
 
 @AWSRetry.exponential_backoff(max_delay=120)
-def create_bucket(s3_client, bucket_name, location):
+def create_bucket(s3_client, bucket_name, location, object_lock_enabled=False):
     try:
+        params = {"Bucket": bucket_name}
+
         configuration = {}
         if location not in ('us-east-1', None):
             configuration['LocationConstraint'] = location
-        if len(configuration) > 0:
-            s3_client.create_bucket(Bucket=bucket_name, CreateBucketConfiguration=configuration)
-        else:
-            s3_client.create_bucket(Bucket=bucket_name)
+
+        if configuration:
+            params["CreateBucketConfiguration"] = configuration
+
+        if object_lock_enabled is not None:
+            params["ObjectLockEnabledForBucket"] = object_lock_enabled
+
+        s3_client.create_bucket(**params)
+
         return True
     except is_boto3_error_code('BucketAlreadyOwnedByYou'):
         # We should never get here since we check the bucket presence before calling the create_or_update_bucket
@@ -722,6 +762,12 @@ def put_bucket_versioning(s3_client, bucket_name, required_versioning):
     s3_client.put_bucket_versioning(Bucket=bucket_name, VersioningConfiguration={'Status': required_versioning})
 
 
+@AWSRetry.exponential_backoff(max_delay=120, catch_extra_error_codes=["NoSuchBucket", "OperationAborted"])
+def get_bucket_object_lock_enabled(s3_client, bucket_name):
+    object_lock_configuration = s3_client.get_object_lock_configuration(Bucket=bucket_name)
+    return object_lock_configuration["ObjectLockConfiguration"]["ObjectLockEnabled"] == "Enabled"
+
+
 @AWSRetry.exponential_backoff(max_delay=120, catch_extra_error_codes=['NoSuchBucket', 'OperationAborted'])
 def get_bucket_encryption(s3_client, bucket_name):
     try:
@@ -1111,6 +1157,7 @@ def main():
         delete_object_ownership=dict(type='bool', default=False),
         acl=dict(type='str', choices=['private', 'public-read', 'public-read-write', 'authenticated-read']),
         validate_bucket_name=dict(type='bool', default=True),
+        object_lock_enabled=dict(type="bool"),
     )
 
     required_by = dict(
diff --git a/tests/integration/targets/s3_bucket/inventory b/tests/integration/targets/s3_bucket/inventory
index b79b5d6cc73..db31e4a9b37 100644
--- a/tests/integration/targets/s3_bucket/inventory
+++ b/tests/integration/targets/s3_bucket/inventory
@@ -10,6 +10,7 @@ encryption_bucket_key
 encryption_sse
 public_access
 acl
+object_lock
 
 [all:vars]
 ansible_connection=local
diff --git a/tests/integration/targets/s3_bucket/roles/s3_bucket/tasks/object_lock.yml b/tests/integration/targets/s3_bucket/roles/s3_bucket/tasks/object_lock.yml
new file mode 100644
index 00000000000..9140a566b96
--- /dev/null
+++ b/tests/integration/targets/s3_bucket/roles/s3_bucket/tasks/object_lock.yml
@@ -0,0 +1,131 @@
+---
+- module_defaults:
+    group/aws:
+      aws_access_key: "{{ aws_access_key }}"
+      aws_secret_key: "{{ aws_secret_key }}"
+      security_token: "{{ security_token | default(omit) }}"
+      region: "{{ aws_region }}"
+  block:
+    - set_fact:
+        local_bucket_name: "{{ bucket_name | hash('md5')}}-objectlock"
+
+    # ============================================================
+
+    - name: 'Create a simple bucket'
+      s3_bucket:
+        name: '{{ local_bucket_name }}'
+        state: present
+      register: output
+
+    - assert:
+        that:
+          - output.changed
+          - not output.object_lock_enabled
+
+    - name: 'Re-disable object lock (idempotency)'
+      s3_bucket:
+        name: '{{ local_bucket_name }}'
+        state: present
+        object_lock_enabled: false
+      register: output
+
+    - assert:
+        that:
+          - not output.changed
+          - not output.object_lock_enabled
+
+    - name: 'Enable object lock'
+      s3_bucket:
+        name: '{{ local_bucket_name }}'
+        state: present
+        object_lock_enabled: true
+      register: output
+      ignore_errors: true
+
+    - assert:
+        that:
+          - output is failed
+
+    - name: Delete test s3 bucket
+      s3_bucket:
+        name: '{{ local_bucket_name }}'
+        state: absent
+      register: output
+
+    - assert:
+        that:
+          - output.changed
+
+    # ============================================================
+
+    - name: 'Create a bucket with object lock enabled'
+      s3_bucket:
+        name: '{{ local_bucket_name }}-2'
+        state: present
+        object_lock_enabled: true
+      register: output
+
+    - assert:
+        that:
+          - output.changed
+          - output.object_lock_enabled
+
+    - name: 'Disable object lock'
+      s3_bucket:
+        name: '{{ local_bucket_name }}-2'
+        state: present
+        object_lock_enabled: false
+      register: output
+      ignore_errors: true
+
+    - assert:
+        that:
+          - output is failed
+
+    - name: 'Re-Enable object lock (idempotency)'
+      s3_bucket:
+        name: '{{ local_bucket_name }}-2'
+        state: present
+        object_lock_enabled: true
+      register: output
+
+    - assert:
+        that:
+          - not output.changed
+          - output.object_lock_enabled
+
+    - name: 'Touch bucket with object lock enabled (idempotency)'
+      s3_bucket:
+        name: '{{ local_bucket_name }}-2'
+        state: present
+        object_lock_enabled: true
+      register: output
+
+    - assert:
+        that:
+          - not output.changed
+          - output.object_lock_enabled
+
+    - name: Delete test s3 bucket
+      s3_bucket:
+        name: '{{ local_bucket_name }}-2'
+        state: absent
+      register: output
+
+    - assert:
+        that:
+          - output.changed
+
+  # ============================================================
+  always:
+    - name: Ensure all buckets are deleted
+      s3_bucket:
+        name: '{{ local_bucket_name }}'
+        state: absent
+      ignore_errors: yes
+
+    - name: Ensure all buckets are deleted
+      s3_bucket:
+        name: '{{ local_bucket_name }}-2'
+        state: absent
+      ignore_errors: yes