diff --git a/moto/s3/models.py b/moto/s3/models.py
index 1db77064f..dd49550a6 100644
--- a/moto/s3/models.py
+++ b/moto/s3/models.py
@@ -1477,6 +1477,28 @@ class S3Backend(BaseBackend):
     def delete_bucket_encryption(self, bucket_name):
         self.get_bucket(bucket_name).encryption = None
 
+    def get_bucket_replication(self, bucket_name):
+        bucket = self.get_bucket(bucket_name)
+        return getattr(bucket, "replication", None)
+
+    def put_bucket_replication(self, bucket_name, replication):
+        if isinstance(replication["Rule"], dict):
+            replication["Rule"] = [replication["Rule"]]
+        for rule in replication["Rule"]:
+            if "Priority" not in rule:
+                rule["Priority"] = 1
+            if "ID" not in rule:
+                rule["ID"] = "".join(
+                    random.choice(string.ascii_letters + string.digits)
+                    for _ in range(30)
+                )
+        bucket = self.get_bucket(bucket_name)
+        bucket.replication = replication
+
+    def delete_bucket_replication(self, bucket_name):
+        bucket = self.get_bucket(bucket_name)
+        bucket.replication = None
+
     def put_bucket_lifecycle(self, bucket_name, rules):
         bucket = self.get_bucket(bucket_name)
         bucket.set_lifecycle(rules)
diff --git a/moto/s3/responses.py b/moto/s3/responses.py
index 7904253dc..73115d32c 100644
--- a/moto/s3/responses.py
+++ b/moto/s3/responses.py
@@ -513,6 +513,13 @@ class ResponseObject(_TemplateEnvironmentMixin, ActionAuthenticatorMixin):
             return 200, {}, template.render(encryption=encryption)
         elif querystring.get("list-type", [None])[0] == "2":
             return 200, {}, self._handle_list_objects_v2(bucket_name, querystring)
+        elif "replication" in querystring:
+            replication = self.backend.get_bucket_replication(bucket_name)
+            if not replication:
+                template = self.response_template(S3_NO_REPLICATION)
+                return 404, {}, template.render(bucket_name=bucket_name)
+            template = self.response_template(S3_REPLICATION_CONFIG)
+            return 200, {}, template.render(replication=replication)
 
         bucket = self.backend.get_bucket(bucket_name)
         prefix = querystring.get("prefix", [None])[0]
@@ -781,6 +788,14 @@ class ResponseObject(_TemplateEnvironmentMixin, ActionAuthenticatorMixin):
                 raise MalformedXML()
             except Exception as e:
                 raise e
+        elif "replication" in querystring:
+            bucket = self.backend.get_bucket(bucket_name)
+            if not bucket.is_versioned:
+                template = self.response_template(S3_NO_VERSIONING_ENABLED)
+                return 400, {}, template.render(bucket_name=bucket_name)
+            replication_config = self._replication_config_from_xml(body)
+            self.backend.put_bucket_replication(bucket_name, replication_config)
+            return ""
         else:
             # us-east-1, the default AWS region behaves a bit differently
             # - you should not use it as a location constraint --> it fails
@@ -865,6 +880,9 @@ class ResponseObject(_TemplateEnvironmentMixin, ActionAuthenticatorMixin):
         elif "encryption" in querystring:
             self.backend.delete_bucket_encryption(bucket_name)
             return 204, {}, ""
+        elif "replication" in querystring:
+            self.backend.delete_bucket_replication(bucket_name)
+            return 204, {}, ""
 
         removed_bucket = self.backend.delete_bucket(bucket_name)
 
@@ -1893,6 +1911,11 @@ class ResponseObject(_TemplateEnvironmentMixin, ActionAuthenticatorMixin):
         config = parsed_xml["AccelerateConfiguration"]
         return config["Status"]
 
+    def _replication_config_from_xml(self, xml):
+        parsed_xml = xmltodict.parse(xml, dict_constructor=dict)
+        config = parsed_xml["ReplicationConfiguration"]
+        return config
+
     def _key_response_delete(self, headers, bucket_name, query, key_name):
         self._set_action("KEY", "DELETE", query)
         self._authenticate_and_authorize_s3_action()
@@ -2707,3 +2730,45 @@ S3_DUPLICATE_BUCKET_ERROR = """<?xml version="1.0" encoding="UTF-8"?>
   <HostId>9Gjjt1m+cjU4OPvX9O9/8RuvnG41MRb/18Oux2o5H5MY7ISNTlXN+Dz9IG62/ILVxhAGI0qyPfg=</HostId>
 </Error>
 """
+
+S3_NO_REPLICATION = """<?xml version="1.0" encoding="UTF-8"?>
+<Error>
+  <Code>ReplicationConfigurationNotFoundError</Code>
+  <Message>The replication configuration was not found</Message>
+  <BucketName>{{ bucket_name }}</BucketName>
+  <RequestId>ZM6MA8EGCZ1M9EW9</RequestId>
+  <HostId>SMUZFedx1CuwjSaZQnM2bEVpet8UgX9uD/L7e MlldClgtEICTTVFz3C66cz8Bssci2OsWCVlog=</HostId>
+</Error>
+"""
+
+S3_NO_VERSIONING_ENABLED = """<?xml version="1.0" encoding="UTF-8"?>
+<Error>
+  <Code>InvalidRequest</Code>
+  <Message>Versioning must be 'Enabled' on the bucket to apply a replication configuration</Message>
+  <BucketName>{{ bucket_name }}</BucketName>
+  <RequestId>ZM6MA8EGCZ1M9EW9</RequestId>
+  <HostId>SMUZFedx1CuwjSaZQnM2bEVpet8UgX9uD/L7e MlldClgtEICTTVFz3C66cz8Bssci2OsWCVlog=</HostId>
+</Error>
+"""
+
+S3_REPLICATION_CONFIG = """<?xml version="1.0" encoding="UTF-8"?>
+<ReplicationConfiguration xmlns="http://s3.amazonaws.com/doc/2006-03-01/">
+{% for rule in replication["Rule"] %}
+<Rule>
+  <ID>{{ rule["ID"] }}</ID>
+  <Priority>{{ rule["Priority"] }}</Priority>
+  <Status>{{ rule["Status"] }}</Status>
+  <DeleteMarkerReplication>
+    <Status>Disabled</Status>
+  </DeleteMarkerReplication>
+  <Filter>
+    <Prefix></Prefix>
+  </Filter>
+  <Destination>
+    <Bucket>{{ rule["Destination"]["Bucket"] }}</Bucket>
+  </Destination>
+</Rule>
+{% endfor %}
+<Role>{{ replication["Role"] }}</Role>
+</ReplicationConfiguration>
+"""
diff --git a/tests/test_s3/test_s3_replication.py b/tests/test_s3/test_s3_replication.py
new file mode 100644
index 000000000..442e63706
--- /dev/null
+++ b/tests/test_s3/test_s3_replication.py
@@ -0,0 +1,176 @@
+import boto3
+import pytest
+import sure  # noqa
+
+from botocore.exceptions import ClientError
+from moto import mock_s3
+from uuid import uuid4
+
+DEFAULT_REGION_NAME = "us-east-1"
+
+
+@mock_s3
+def test_get_bucket_replication_for_unexisting_bucket():
+    bucket_name = str(uuid4())
+    s3 = boto3.client("s3", region_name=DEFAULT_REGION_NAME)
+    with pytest.raises(ClientError) as exc:
+        s3.get_bucket_replication(Bucket=bucket_name)
+    err = exc.value.response["Error"]
+    err["Code"].should.equal("NoSuchBucket")
+    err["Message"].should.equal("The specified bucket does not exist")
+    err["BucketName"].should.equal(bucket_name)
+
+
+@mock_s3
+def test_get_bucket_replication_bucket_without_replication():
+    bucket_name = str(uuid4())
+    s3 = boto3.client("s3", region_name=DEFAULT_REGION_NAME)
+    s3.create_bucket(Bucket=bucket_name)
+
+    with pytest.raises(ClientError) as exc:
+        s3.get_bucket_replication(Bucket=bucket_name)
+    err = exc.value.response["Error"]
+    err["Code"].should.equal("ReplicationConfigurationNotFoundError")
+    err["Message"].should.equal("The replication configuration was not found")
+    err["BucketName"].should.equal(bucket_name)
+
+
+@mock_s3
+def test_delete_bucket_replication_unknown_bucket():
+    bucket_name = str(uuid4())
+    s3 = boto3.client("s3", region_name=DEFAULT_REGION_NAME)
+    with pytest.raises(ClientError) as exc:
+        s3.delete_bucket_replication(Bucket=bucket_name)
+    err = exc.value.response["Error"]
+    err["Code"].should.equal("NoSuchBucket")
+    err["Message"].should.equal("The specified bucket does not exist")
+    err["BucketName"].should.equal(bucket_name)
+
+
+@mock_s3
+def test_delete_bucket_replication_bucket_without_replication():
+    bucket_name = str(uuid4())
+    s3 = boto3.client("s3", region_name=DEFAULT_REGION_NAME)
+
+    s3.create_bucket(Bucket=bucket_name)
+    # No-op
+    s3.delete_bucket_replication(Bucket=bucket_name)
+
+
+@mock_s3
+def test_create_replication_without_versioning():
+    bucket_name = str(uuid4())
+    s3 = boto3.client("s3", region_name=DEFAULT_REGION_NAME)
+    s3.create_bucket(Bucket=bucket_name)
+
+    with pytest.raises(ClientError) as exc:
+        s3.put_bucket_replication(
+            Bucket=bucket_name,
+            ReplicationConfiguration={
+                "Role": "myrole",
+                "Rules": [
+                    {"Destination": {"Bucket": "secondbucket"}, "Status": "Enabled"}
+                ],
+            },
+        )
+    err = exc.value.response["Error"]
+    err["Code"].should.equal("InvalidRequest")
+    err["Message"].should.equal(
+        "Versioning must be 'Enabled' on the bucket to apply a replication configuration"
+    )
+    err["BucketName"].should.equal(bucket_name)
+
+
+@mock_s3
+def test_create_and_retrieve_replication_with_single_rules():
+    bucket_name = str(uuid4())
+    s3 = boto3.client("s3", region_name=DEFAULT_REGION_NAME)
+
+    s3.create_bucket(Bucket=bucket_name)
+    s3.put_bucket_versioning(
+        Bucket=bucket_name, VersioningConfiguration={"Status": "Enabled"}
+    )
+    s3.put_bucket_replication(
+        Bucket=bucket_name,
+        ReplicationConfiguration={
+            "Role": "myrole",
+            "Rules": [
+                {
+                    "ID": "firstrule",
+                    "Priority": 2,
+                    "Destination": {"Bucket": "secondbucket"},
+                    "Status": "Enabled",
+                }
+            ],
+        },
+    )
+
+    config = s3.get_bucket_replication(Bucket=bucket_name)["ReplicationConfiguration"]
+    config.should.equal(
+        {
+            "Role": "myrole",
+            "Rules": [
+                {
+                    "DeleteMarkerReplication": {"Status": "Disabled"},
+                    "Destination": {"Bucket": "secondbucket"},
+                    "Filter": {"Prefix": ""},
+                    "ID": "firstrule",
+                    "Priority": 2,
+                    "Status": "Enabled",
+                }
+            ],
+        }
+    )
+
+    s3.delete_bucket_replication(Bucket=bucket_name)
+
+    # Can't retrieve replication that has been deleted
+    with pytest.raises(ClientError) as exc:
+        s3.get_bucket_replication(Bucket=bucket_name)
+    err = exc.value.response["Error"]
+    err["Code"].should.equal("ReplicationConfigurationNotFoundError")
+    err["Message"].should.equal("The replication configuration was not found")
+    err["BucketName"].should.equal(bucket_name)
+
+
+@mock_s3
+def test_create_and_retrieve_replication_with_multiple_rules():
+    bucket_name = str(uuid4())
+    s3 = boto3.client("s3", region_name=DEFAULT_REGION_NAME)
+
+    s3.create_bucket(Bucket=bucket_name)
+    s3.put_bucket_versioning(
+        Bucket=bucket_name, VersioningConfiguration={"Status": "Enabled"}
+    )
+    s3.put_bucket_replication(
+        Bucket=bucket_name,
+        ReplicationConfiguration={
+            "Role": "myrole",
+            "Rules": [
+                {"Destination": {"Bucket": "secondbucket"}, "Status": "Enabled"},
+                {
+                    "ID": "secondrule",
+                    "Priority": 2,
+                    "Destination": {"Bucket": "thirdbucket"},
+                    "Status": "Disabled",
+                },
+            ],
+        },
+    )
+
+    config = s3.get_bucket_replication(Bucket=bucket_name)["ReplicationConfiguration"]
+    config.should.have.key("Role").equal("myrole")
+    rules = config["Rules"]
+    rules.should.have.length_of(2)
+
+    first_rule = rules[0]
+    first_rule.should.have.key("ID")
+    first_rule.should.have.key("Priority").equal(1)
+    first_rule.should.have.key("Status").equal("Enabled")
+    first_rule.should.have.key("Destination").equal({"Bucket": "secondbucket"})
+
+    second = rules[1]
+    second.should.have.key("ID").equal("secondrule")
+    second.should.have.key("Priority").equal(2)
+    second.should.have.key("Status").equal("Disabled")
+    second.should.have.key("Destination").equal({"Bucket": "thirdbucket"})