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..b3f9efdb543
--- /dev/null
+++ b/changelogs/fragments/1347-s3-object-lock-enabled.yml
@@ -0,0 +1,2 @@
+minor_changes:
+- s3_bucket - The option to create a bucket 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 73d9e6649c4..f7dc2f1f380 100644
--- a/plugins/modules/s3_bucket.py
+++ b/plugins/modules/s3_bucket.py
@@ -163,6 +163,12 @@
     type: bool
     default: false
     version_added: 6.0.0
+  object_lock_enabled:
+    description:
+      - Whether S3 Object Lock to be enabled
+    type: bool
+    default: false
+    version_added: 6.0.0
 
 extends_documentation_fragment:
   - amazon.aws.common.modules
@@ -384,6 +390,7 @@ def create_or_update_bucket(s3_client, module):
     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")
     # default to US Standard region,
     # note: module.region will also try to pull a default out of the boto3 configs.
@@ -401,7 +408,7 @@ def create_or_update_bucket(s3_client, module):
 
     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:
@@ -663,15 +670,15 @@ 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:
         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)
+            s3_client.create_bucket(Bucket=bucket_name, CreateBucketConfiguration=configuration, ObjectLockEnabledForBucket=object_lock_enabled)
         else:
-            s3_client.create_bucket(Bucket=bucket_name)
+            s3_client.create_bucket(Bucket=bucket_name, ObjectLockEnabledForBucket=object_lock_enabled)
         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
@@ -1085,6 +1092,7 @@ def main():
         acl=dict(type='str', choices=['private', 'public-read', 'public-read-write', 'authenticated-read']),
         validate_bucket_name=dict(type='bool', default=True),
         dualstack=dict(default=False, type="bool"),
+        object_lock_enabled=dict(type='bool', default=False),
     )
 
     required_by = dict(
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..7713e048232
--- /dev/null
+++ b/tests/integration/targets/s3_bucket/roles/s3_bucket/tasks/object_lock.yml
@@ -0,0 +1,81 @@
+---
+- 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')}}.object_lock"
+    # ============================================================
+
+    - name: 'Create a simple bucket'
+      s3_bucket:
+        name: '{{ local_bucket_name }}'
+        state: present
+      register: output
+
+    - assert:
+        that:
+          - output.changed
+          - output.object_lock_enabled == False
+
+    # - name: 'Re-enable AES256 encryption (idempotency)'
+    #   s3_bucket:
+    #     name: '{{ local_bucket_name }}'
+    #     state: present
+    #     encryption: 'AES256'
+    #   register: output
+
+    # - assert:
+    #     that:
+    #       - not output.changed
+    #       - output.encryption
+    #       - output.encryption.SSEAlgorithm == 'AES256'
+
+    # # ============================================================
+
+    # - name: Disable encryption from bucket
+    #   s3_bucket:
+    #     name: '{{ local_bucket_name }}'
+    #     state: present
+    #     encryption: "none"
+    #   register: output
+
+    # - assert:
+    #     that:
+    #       - output.changed
+    #       - not output.encryption
+
+    # - name: Disable encryption from bucket
+    #   s3_bucket:
+    #     name: '{{ local_bucket_name }}'
+    #     state: present
+    #     encryption: "none"
+    #   register: output
+
+    # - assert:
+    #     that:
+    #       - output is not changed
+    #       - not output.encryption
+
+    # ============================================================
+
+    - name: Delete encryption test s3 bucket
+      s3_bucket:
+        name: '{{ local_bucket_name }}'
+        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