Repository: getmoto/moto
Pull request: moto/route53 does not correctly handle private hosted zones #4785
models.py:
- Route53Backend.create_hosted_zone now receive vpcid, vpcregion
- Passes into FakeZone.__init___

responses.py:
- Class Route53 now handles the existence of a VPC key in the returned "HOSTEDZONECONFIG
- Cleaned up Route53.list_or_create_hostzone_response() for better readability
- If a CreateHostedZoneRequest comes in without a VPC block we now throw a KeyError
- Added VPC block to GET_HOSTED_ZONE_RESPONSE

test_route53.py:
- import mock_ec2 to create mock VPCs
- test_hosted_zone_private_zone_preserved_boto3()
  - now creates a VPC and a CreateHostedZoneRequest with a VPC block
  - return is now tested:
    - for VPCs section
    - that VPCId and VPCRegion are non-empty
    - that VPCId and VPCRegion match what we created the VPC with and what was sent to the initial CreateHostedZoneRequest
- test_list_hosted_zones_by_name()
  - still creates 3 zones,
  - but 1 is a true Private zone with VPC block
  - others are Public zone by virtue of Private=False
  - return for Private=True zone tested same as test_hosted_zone_private_zone_preserved_boto3()
  - other zones with Private=False set are tested for that.

Problem statement:
moto/route53 does not correctly handle private hosted zones
moto/route53 does not correctly handle private hosted zones with `create_hosted_zone()` such that when calling `list_hosted_zones()` or `get_hosted_zone()` with a specific zone ID, the associated VPC information is not available.

Further, calling `create_hosted_zone()` and requesting a private hosted zone is allowed without passing in an associated VPC configuration block, which is incorrect behavior.

Hints from issue discussion:
I've attempted to fix the above here: https://github.com/spulec/moto/pull/4786

Commit message: Remove unused vars

Patch to review:
diff --git a/moto/route53/models.py b/moto/route53/models.py
index 52c6620ad1f0..a0cd35e7287e 100644
--- a/moto/route53/models.py
+++ b/moto/route53/models.py
@@ -212,11 +212,17 @@ def reverse_domain_name(domain_name):
 
 
 class FakeZone(CloudFormationModel):
-    def __init__(self, name, id_, private_zone, comment=None):
+    def __init__(
+        self, name, id_, private_zone, vpcid=None, vpcregion=None, comment=None
+    ):
         self.name = name
         self.id = id_
         if comment is not None:
             self.comment = comment
+        if vpcid is not None:
+            self.vpcid = vpcid
+        if vpcregion is not None:
+            self.vpcregion = vpcregion
         self.private_zone = private_zone
         self.rrsets = []
 
@@ -365,9 +371,18 @@ def __init__(self):
         self.resource_tags = defaultdict(dict)
         self.query_logging_configs = {}
 
-    def create_hosted_zone(self, name, private_zone, comment=None):
+    def create_hosted_zone(
+        self, name, private_zone, vpcid=None, vpcregion=None, comment=None
+    ):
         new_id = create_route53_zone_id()
-        new_zone = FakeZone(name, new_id, private_zone=private_zone, comment=comment)
+        new_zone = FakeZone(
+            name,
+            new_id,
+            private_zone=private_zone,
+            vpcid=vpcid,
+            vpcregion=vpcregion,
+            comment=comment,
+        )
         self.zones[new_id] = new_zone
         return new_zone
 
diff --git a/moto/route53/responses.py b/moto/route53/responses.py
index daf673c257b8..f7a4093bcbb0 100644
--- a/moto/route53/responses.py
+++ b/moto/route53/responses.py
@@ -30,32 +30,47 @@ class Route53(BaseResponse):
     def list_or_create_hostzone_response(self, request, full_url, headers):
         self.setup_class(request, full_url, headers)
 
+        # Set these here outside the scope of the try/except
+        # so they're defined later when we call create_hosted_zone()
+        vpcid = None
+        vpcregion = None
         if request.method == "POST":
             elements = xmltodict.parse(self.body)
-            if "HostedZoneConfig" in elements["CreateHostedZoneRequest"]:
-                comment = elements["CreateHostedZoneRequest"]["HostedZoneConfig"][
-                    "Comment"
-                ]
+            zone_request = elements["CreateHostedZoneRequest"]
+            if "HostedZoneConfig" in zone_request:
+                zone_config = zone_request["HostedZoneConfig"]
+                comment = zone_config["Comment"]
                 try:
                     # in boto3, this field is set directly in the xml
-                    private_zone = elements["CreateHostedZoneRequest"][
-                        "HostedZoneConfig"
-                    ]["PrivateZone"]
+                    private_zone = zone_config["PrivateZone"]
                 except KeyError:
-                    # if a VPC subsection is only included in xmls params when private_zone=True,
-                    # see boto: boto/route53/connection.py
-                    private_zone = "VPC" in elements["CreateHostedZoneRequest"]
+                    # if a VPC subsection is only included in xmls
+                    # params when private_zone=True, see boto:
+                    # boto/route53/connection.py
+                    private_zone = "VPC" in zone_request
             else:
                 comment = None
                 private_zone = False
 
-            name = elements["CreateHostedZoneRequest"]["Name"]
+            if private_zone == "true":
+                try:
+                    vpcid = zone_request["VPC"]["VPCId"]
+                    vpcregion = zone_request["VPC"]["VPCRegion"]
+                except KeyError:
+                    msg = "Private Hosted zone requires VPC ID and Region"
+                    raise KeyError(msg)
+
+            name = zone_request["Name"]
 
             if name[-1] != ".":
                 name += "."
 
             new_zone = route53_backend.create_hosted_zone(
-                name, comment=comment, private_zone=private_zone
+                name,
+                comment=comment,
+                private_zone=private_zone,
+                vpcid=vpcid,
+                vpcregion=vpcregion,
             )
             template = Template(CREATE_HOSTED_ZONE_RESPONSE)
             return 201, headers, template.render(zone=new_zone)
@@ -414,6 +429,13 @@ def get_or_delete_query_logging_config_response(self, request, full_url, headers
          <NameServer>moto.test.com</NameServer>
       </NameServers>
    </DelegationSet>
+   <VPCs>
+      <VPC>
+         <VPCId>{{zone.vpcid}}</VPCId>
+         <VPCRegion>{{zone.vpcregion}}</VPCRegion>
+      </VPC>
+   </VPCs>
+
 </GetHostedZoneResponse>"""
 
 CREATE_HOSTED_ZONE_RESPONSE = """<CreateHostedZoneResponse xmlns="https://route53.amazonaws.com/doc/2012-12-12/">
@@ -433,6 +455,10 @@ def get_or_delete_query_logging_config_response(self, request, full_url, headers
          <NameServer>moto.test.com</NameServer>
       </NameServers>
    </DelegationSet>
+   <VPC>
+      <VPCId>{{zone.vpcid}}</VPCId>
+      <VPCRegion>{{zone.vpcregion}}</VPCRegion>
+   </VPC>
 </CreateHostedZoneResponse>"""
 
 LIST_HOSTED_ZONES_RESPONSE = """<ListHostedZonesResponse xmlns="https://route53.amazonaws.com/doc/2012-12-12/">
diff --git a/tests/test_route53/test_route53.py b/tests/test_route53/test_route53.py
index 3516b64fd3f3..b5b8d9619c8e 100644
--- a/tests/test_route53/test_route53.py
+++ b/tests/test_route53/test_route53.py
@@ -6,7 +6,7 @@
 import botocore
 import pytest
 
-from moto import mock_route53
+from moto import mock_ec2, mock_route53
 
 
 @mock_route53
@@ -367,21 +367,34 @@ def test_deleting_latency_route_boto3():
     cnames[0]["Region"].should.equal("us-west-1")
 
 
+@mock_ec2
 @mock_route53
 def test_hosted_zone_private_zone_preserved_boto3():
-    conn = boto3.client("route53", region_name="us-east-1")
-    # TODO: actually create_hosted_zone statements with PrivateZone=True, but without
-    # a _valid_ vpc-id should fail.
-    firstzone = conn.create_hosted_zone(
+    # Create mock VPC so we can get a VPC ID
+    region = "us-east-1"
+    ec2c = boto3.client("ec2", region_name=region)
+    vpc_id = ec2c.create_vpc(CidrBlock="10.1.0.0/16").get("Vpc").get("VpcId")
+
+    # Create hosted_zone as a Private VPC Hosted Zone
+    conn = boto3.client("route53", region_name=region)
+    new_zone = conn.create_hosted_zone(
         Name="testdns.aws.com.",
         CallerReference=str(hash("foo")),
         HostedZoneConfig=dict(PrivateZone=True, Comment="Test"),
+        VPC={"VPCRegion": region, "VPCId": vpc_id},
     )
 
-    zone_id = firstzone["HostedZone"]["Id"].split("/")[-1]
-
+    zone_id = new_zone["HostedZone"]["Id"].split("/")[-1]
     hosted_zone = conn.get_hosted_zone(Id=zone_id)
     hosted_zone["HostedZone"]["Config"]["PrivateZone"].should.equal(True)
+    hosted_zone.should.have.key("VPCs")
+    hosted_zone["VPCs"].should.have.length_of(1)
+    hosted_zone["VPCs"][0].should.have.key("VPCId")
+    hosted_zone["VPCs"][0].should.have.key("VPCRegion")
+    hosted_zone["VPCs"][0]["VPCId"].should_not.be.empty
+    hosted_zone["VPCs"][0]["VPCRegion"].should_not.be.empty
+    hosted_zone["VPCs"][0]["VPCId"].should.be.equal(vpc_id)
+    hosted_zone["VPCs"][0]["VPCRegion"].should.be.equal(region)
 
     hosted_zones = conn.list_hosted_zones()
     hosted_zones["HostedZones"][0]["Config"]["PrivateZone"].should.equal(True)
@@ -390,6 +403,21 @@ def test_hosted_zone_private_zone_preserved_boto3():
     len(hosted_zones["HostedZones"]).should.equal(1)
     hosted_zones["HostedZones"][0]["Config"]["PrivateZone"].should.equal(True)
 
+    # create_hosted_zone statements with  PrivateZone=True,
+    # but without a _valid_ vpc-id should fail.
+    conn = boto3.client("route53", region_name=region)
+    with pytest.raises(KeyError) as e:
+        conn.create_hosted_zone(
+            Name="testdns.aws.com.",
+            CallerReference=str(hash("foo")),
+            HostedZoneConfig=dict(PrivateZone=True, Comment="Test"),
+        )
+        print("")
+    expected_msg = "Private Hosted zone requires VPC ID and Region"
+    assert e.value.args[0] == expected_msg
+
+    return
+
 
 @mock_route53
 def test_list_or_change_tags_for_resource_request():
@@ -479,33 +507,77 @@ def test_list_or_change_tags_for_resource_request():
     response["ResourceTagSet"]["Tags"].should.be.empty
 
 
+@mock_ec2
 @mock_route53
 def test_list_hosted_zones_by_name():
-    conn = boto3.client("route53", region_name="us-east-1")
-    conn.create_hosted_zone(
+
+    # Create mock VPC so we can get a VPC ID
+    ec2c = boto3.client("ec2", region_name="us-east-1")
+    vpc_id = ec2c.create_vpc(CidrBlock="10.1.0.0/16").get("Vpc").get("VpcId")
+    region = "us-east-1"
+
+    conn = boto3.client("route53", region_name=region)
+    zone_b = conn.create_hosted_zone(
         Name="test.b.com.",
         CallerReference=str(hash("foo")),
         HostedZoneConfig=dict(PrivateZone=True, Comment="test com"),
+        VPC={"VPCRegion": region, "VPCId": vpc_id},
     )
+
+    zone_b = conn.list_hosted_zones_by_name(DNSName="test.b.com.")
+    len(zone_b["HostedZones"]).should.equal(1)
+    zone_b["HostedZones"][0]["Name"].should.equal("test.b.com.")
+    zone_b["HostedZones"][0].should.have.key("Config")
+    zone_b["HostedZones"][0]["Config"].should.have.key("PrivateZone")
+    zone_b["HostedZones"][0]["Config"]["PrivateZone"].should.be.equal(True)
+
+    # We declared this a a private hosted zone above, so let's make
+    # sure it really is!
+    zone_b_id = zone_b["HostedZones"][0]["Id"].split("/")[-1]
+    b_hosted_zone = conn.get_hosted_zone(Id=zone_b_id)
+
+    # Pull the HostedZone block out and test it.
+    b_hosted_zone.should.have.key("HostedZone")
+    b_hz = b_hosted_zone["HostedZone"]
+    b_hz.should.have.key("Config")
+    b_hz["Config"].should.have.key("PrivateZone")
+    b_hz["Config"]["PrivateZone"].should.be.equal(True)
+
+    # Check for the VPCs block since this *should* be a VPC-Private Zone
+    b_hosted_zone.should.have.key("VPCs")
+    b_hosted_zone["VPCs"].should.have.length_of(1)
+    b_hz_vpcs = b_hosted_zone["VPCs"][0]
+    b_hz_vpcs.should.have.key("VPCId")
+    b_hz_vpcs.should.have.key("VPCRegion")
+    b_hz_vpcs["VPCId"].should_not.be.empty
+    b_hz_vpcs["VPCRegion"].should_not.be.empty
+    b_hz_vpcs["VPCId"].should.be.equal(vpc_id)
+    b_hz_vpcs["VPCRegion"].should.be.equal(region)
+
+    # Now create other zones and test them.
     conn.create_hosted_zone(
         Name="test.a.org.",
         CallerReference=str(hash("bar")),
-        HostedZoneConfig=dict(PrivateZone=True, Comment="test org"),
+        HostedZoneConfig=dict(PrivateZone=False, Comment="test org"),
     )
     conn.create_hosted_zone(
         Name="test.a.org.",
         CallerReference=str(hash("bar")),
-        HostedZoneConfig=dict(PrivateZone=True, Comment="test org 2"),
+        HostedZoneConfig=dict(PrivateZone=False, Comment="test org 2"),
     )
 
-    # test lookup
-    zones = conn.list_hosted_zones_by_name(DNSName="test.b.com.")
-    len(zones["HostedZones"]).should.equal(1)
-    zones["HostedZones"][0]["Name"].should.equal("test.b.com.")
+    # Now makes sure the other zones we created above are NOT private...
     zones = conn.list_hosted_zones_by_name(DNSName="test.a.org.")
     len(zones["HostedZones"]).should.equal(2)
     zones["HostedZones"][0]["Name"].should.equal("test.a.org.")
+    zones["HostedZones"][0].should.have.key("Config")
+    zones["HostedZones"][0]["Config"].should.have.key("PrivateZone")
+    zones["HostedZones"][0]["Config"]["PrivateZone"].should.be.equal(False)
+
     zones["HostedZones"][1]["Name"].should.equal("test.a.org.")
+    zones["HostedZones"][1].should.have.key("Config")
+    zones["HostedZones"][1]["Config"].should.have.key("PrivateZone")
+    zones["HostedZones"][1]["Config"]["PrivateZone"].should.be.equal(False)
 
     # test sort order
     zones = conn.list_hosted_zones_by_name()
@@ -521,17 +593,17 @@ def test_list_hosted_zones_by_dns_name():
     conn.create_hosted_zone(
         Name="test.b.com.",
         CallerReference=str(hash("foo")),
-        HostedZoneConfig=dict(PrivateZone=True, Comment="test com"),
+        HostedZoneConfig=dict(PrivateZone=False, Comment="test com"),
     )
     conn.create_hosted_zone(
         Name="test.a.org.",
         CallerReference=str(hash("bar")),
-        HostedZoneConfig=dict(PrivateZone=True, Comment="test org"),
+        HostedZoneConfig=dict(PrivateZone=False, Comment="test org"),
     )
     conn.create_hosted_zone(
         Name="test.a.org.",
         CallerReference=str(hash("bar")),
-        HostedZoneConfig=dict(PrivateZone=True, Comment="test org 2"),
+        HostedZoneConfig=dict(PrivateZone=False, Comment="test org 2"),
     )
     conn.create_hosted_zone(
         Name="my.test.net.",
@@ -569,7 +641,7 @@ def test_change_resource_record_sets_crud_valid():
     conn.create_hosted_zone(
         Name="db.",
         CallerReference=str(hash("foo")),
-        HostedZoneConfig=dict(PrivateZone=True, Comment="db"),
+        HostedZoneConfig=dict(PrivateZone=False, Comment="db"),
     )
 
     zones = conn.list_hosted_zones_by_name(DNSName="db.")
@@ -707,7 +779,7 @@ def test_change_resource_record_sets_crud_valid_with_special_xml_chars():
     conn.create_hosted_zone(
         Name="db.",
         CallerReference=str(hash("foo")),
-        HostedZoneConfig=dict(PrivateZone=True, Comment="db"),
+        HostedZoneConfig=dict(PrivateZone=False, Comment="db"),
     )
 
     zones = conn.list_hosted_zones_by_name(DNSName="db.")
@@ -977,7 +1049,7 @@ def test_change_resource_record_invalid():
     conn.create_hosted_zone(
         Name="db.",
         CallerReference=str(hash("foo")),
-        HostedZoneConfig=dict(PrivateZone=True, Comment="db"),
+        HostedZoneConfig=dict(PrivateZone=False, Comment="db"),
     )
 
     zones = conn.list_hosted_zones_by_name(DNSName="db.")
@@ -1038,7 +1110,7 @@ def test_list_resource_record_sets_name_type_filters():
     create_hosted_zone_response = conn.create_hosted_zone(
         Name="db.",
         CallerReference=str(hash("foo")),
-        HostedZoneConfig=dict(PrivateZone=True, Comment="db"),
+        HostedZoneConfig=dict(PrivateZone=False, Comment="db"),
     )
     hosted_zone_id = create_hosted_zone_response["HostedZone"]["Id"]
 
@@ -1107,7 +1179,7 @@ def test_change_resource_record_sets_records_limit():
     conn.create_hosted_zone(
         Name="db.",
         CallerReference=str(hash("foo")),
-        HostedZoneConfig=dict(PrivateZone=True, Comment="db"),
+        HostedZoneConfig=dict(PrivateZone=False, Comment="db"),
     )
 
     zones = conn.list_hosted_zones_by_name(DNSName="db.")


Code review comments:
diff_hunk: @@ -390,6 +403,21 @@ def test_hosted_zone_private_zone_preserved_boto3():
     len(hosted_zones["HostedZones"]).should.equal(1)
     hosted_zones["HostedZones"][0]["Config"]["PrivateZone"].should.equal(True)
 
+    # create_hosted_zone statements with  PrivateZone=True,
+    # but without a _valid_ vpc-id should fail.
+    conn = boto3.client("route53", region_name=region)
+    with pytest.raises(KeyError) as e:
path: tests/test_route53/test_route53.py
text: @user1:
Moto in ServerMode will only throw exceptions of type `ClientError`, so we should catch that instead.
The assertion then becomes something like `e.value.response["Error"]["Message"] == expected_msg`

Changing this should get the tests to pass

@author:
Oh, okay.  I'm not sure how to test that locally on my laptop. I'll give it a whirl. Do I just fire up the moto_server and then just run the server tests ?

@user1:
Calling `moto_server` to start it, yes. Then you can use `TEST_SERVER_MODE=true pytest tests/test_route53` to run the same tests against that server.

@author:
Awesome, that's what I suspected. Looks like the Makefile has a `test_server` target as well, so I'm assuming I just need to fire it up on port 5000 (assuming the developer FAQ is current).

I'll see what I can do tonight.

diff_hunk: @@ -30,32 +30,47 @@ class Route53(BaseResponse):
     def list_or_create_hostzone_response(self, request, full_url, headers):
         self.setup_class(request, full_url, headers)
 
+        # Set these here outside the scope of the try/except
+        # so they're defined later when we call create_hosted_zone()
+        vpcid = None
+        vpcregion = None
         if request.method == "POST":
             elements = xmltodict.parse(self.body)
-            if "HostedZoneConfig" in elements["CreateHostedZoneRequest"]:
-                comment = elements["CreateHostedZoneRequest"]["HostedZoneConfig"][
-                    "Comment"
-                ]
+            zone_request = elements["CreateHostedZoneRequest"]
+            if "HostedZoneConfig" in zone_request:
+                zone_config = zone_request["HostedZoneConfig"]
+                comment = zone_config["Comment"]
                 try:
                     # in boto3, this field is set directly in the xml
-                    private_zone = elements["CreateHostedZoneRequest"][
-                        "HostedZoneConfig"
-                    ]["PrivateZone"]
+                    private_zone = zone_config["PrivateZone"]
path: moto/route53/responses.py
text: @user1:
```suggestion
                    private_zone = zone_config["PrivateZone"].lower() == "true"
```
Ideally this returns a boolean - as the `except`-clause also returns a boolean.