Yasuke/moto
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

moto/route53 does not correctly handle private hosted zones #4785 (#4786)

Browse Source
This commit is contained in:
Paul 2022-01-27 06:28:31 -05:00 committed by GitHub
parent 4bd8f4f96f
commit cd2d7a9c7a
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
4 changed files with 166 additions and 51 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

11
moto/route53/exceptions.py
View File

@ -33,6 +33,17 @@ class InvalidPaginationToken(Route53ClientError):
super().__init__("InvalidPaginationToken", message)
class InvalidVPCId(Route53ClientError):
"""Missing/Invalid VPC ID"""
code = 400
def __init__(self):
message = "Invalid or missing VPC Id."
super().__init__("InvalidVPCId", message)
self.content_type = "text/xml"
class NoSuchCloudWatchLogsLogGroup(Route53ClientError):
"""CloudWatch LogGroup has a permissions policy, but does not exist."""

21
moto/route53/models.py
View File

@ -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 @@ class Route53Backend(BaseBackend):
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

50
moto/route53/responses.py
View File

@ -6,7 +6,7 @@ from jinja2 import Template
import xmltodict
from moto.core.responses import BaseResponse
from moto.route53.exceptions import Route53ClientError, InvalidChangeBatch
from moto.route53.exceptions import Route53ClientError, InvalidChangeBatch, InvalidVPCId
from moto.route53.models import route53_backend
XMLNS = "https://route53.amazonaws.com/doc/2013-04-01/"
@ -30,32 +30,39 @@ 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"
]
try:
# in boto3, this field is set directly in the xml
private_zone = elements["CreateHostedZoneRequest"][
"HostedZoneConfig"
]["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"]
zone_request = elements["CreateHostedZoneRequest"]
if "HostedZoneConfig" in zone_request:
zone_config = zone_request["HostedZoneConfig"]
comment = zone_config["Comment"]
private_zone = zone_config.get("PrivateZone", False)
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:
raise InvalidVPCId()
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 +421,13 @@ GET_HOSTED_ZONE_RESPONSE = """<GetHostedZoneResponse xmlns="https://route53.amaz
<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 +447,10 @@ CREATE_HOSTED_ZONE_RESPONSE = """<CreateHostedZoneResponse xmlns="https://route5
<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/">

135
tests/test_route53/test_route53.py
View File

@ -6,7 +6,7 @@ import sure # noqa # pylint: disable=unused-import
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(ClientError) as exc:
conn.create_hosted_zone(
Name="testdns.aws.com.",
CallerReference=str(hash("foo")),
HostedZoneConfig=dict(PrivateZone=True, Comment="Test"),
)
err = exc.value.response["Error"]
err["Code"].should.equal("InvalidVPCId")
err["Message"].should.equal("Invalid or missing VPC Id.")
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"),
)
conn.create_hosted_zone(
Name="test.a.org.",
CallerReference=str(hash("bar")),
HostedZoneConfig=dict(PrivateZone=True, Comment="test org"),
)
conn.create_hosted_zone(
Name="test.a.org.",
CallerReference=str(hash("bar")),
HostedZoneConfig=dict(PrivateZone=True, Comment="test org 2"),
VPC={"VPCRegion": region, "VPCId": vpc_id},
)
# 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.")
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=False, Comment="test org"),
)
conn.create_hosted_zone(
Name="test.a.org.",
CallerReference=str(hash("bar")),
HostedZoneConfig=dict(PrivateZone=False, Comment="test org 2"),
)
# 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.")
@ -1159,7 +1231,6 @@ def test_change_resource_record_sets_records_limit():
"Comment": "Create four records with 250 resource records each, plus one more",
"Changes": too_many_changes,
}
with pytest.raises(ClientError) as exc:
conn.change_resource_record_sets(
HostedZoneId=hosted_zone_id,