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

Route53 - Reusable Delegation Sets (#4817)

Browse Source
This commit is contained in:
Bert Blommers 2022-01-30 23:53:05 -01:00 committed by GitHub
parent 2fd6f34060
commit 0a1bb6bae1
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
8 changed files with 417 additions and 21 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

12
IMPLEMENTATION_COVERAGE.md
View File

@ -4110,7 +4110,7 @@
## route53
<details>
<summary>26% implemented</summary>
<summary>34% implemented</summary>
- [ ] activate_key_signing_key
- [ ] associate_vpc_with_hosted_zone
@ -4120,7 +4120,7 @@
- [X] create_hosted_zone
- [ ] create_key_signing_key
- [X] create_query_logging_config
- [ ] create_reusable_delegation_set
- [X] create_reusable_delegation_set
- [ ] create_traffic_policy
- [ ] create_traffic_policy_instance
- [ ] create_traffic_policy_version
@ -4130,7 +4130,7 @@
- [X] delete_hosted_zone
- [ ] delete_key_signing_key
- [X] delete_query_logging_config
- [ ] delete_reusable_delegation_set
- [X] delete_reusable_delegation_set
- [ ] delete_traffic_policy
- [ ] delete_traffic_policy_instance
- [ ] delete_vpc_association_authorization
@ -4147,10 +4147,10 @@
- [ ] get_health_check_last_failure_reason
- [ ] get_health_check_status
- [X] get_hosted_zone
- [ ] get_hosted_zone_count
- [X] get_hosted_zone_count
- [ ] get_hosted_zone_limit
- [X] get_query_logging_config
- [ ] get_reusable_delegation_set
- [X] get_reusable_delegation_set
- [ ] get_reusable_delegation_set_limit
- [ ] get_traffic_policy
- [ ] get_traffic_policy_instance
@ -4162,7 +4162,7 @@
- [X] list_hosted_zones_by_vpc
- [X] list_query_logging_configs
- [X] list_resource_record_sets
- [ ] list_reusable_delegation_sets
- [X] list_reusable_delegation_sets
- [X] list_tags_for_resource
- [ ] list_tags_for_resources
- [ ] list_traffic_policies

16
docs/docs/services/route53.rst
View File

@ -35,7 +35,7 @@ route53
- [X] create_query_logging_config
Process the create_query_logging_config request.
- [ ] create_reusable_delegation_set
- [X] create_reusable_delegation_set
- [ ] create_traffic_policy
- [ ] create_traffic_policy_instance
- [ ] create_traffic_policy_version
@ -47,7 +47,7 @@ route53
- [X] delete_query_logging_config
Delete query logging config, if it exists.
- [ ] delete_reusable_delegation_set
- [X] delete_reusable_delegation_set
- [ ] delete_traffic_policy
- [ ] delete_traffic_policy_instance
- [ ] delete_vpc_association_authorization
@ -69,7 +69,7 @@ route53
- [X] get_query_logging_config
Return query logging config, if it exists.
- [ ] get_reusable_delegation_set
- [X] get_reusable_delegation_set
- [ ] get_reusable_delegation_set_limit
- [ ] get_traffic_policy
- [ ] get_traffic_policy_instance
@ -83,7 +83,15 @@ route53
Return a list of query logging configs.
- [X] list_resource_record_sets
- [ ] list_reusable_delegation_sets
The StartRecordIdentifier-parameter is not yet implemented
- [X] list_reusable_delegation_sets
Pagination is not yet implemented
- [X] list_tags_for_resource
- [ ] list_tags_for_resources
- [ ] list_traffic_policies

8
moto/route53/exceptions.py
View File

@ -92,3 +92,11 @@ class InvalidChangeBatch(Route53ClientError):
def __init__(self):
message = "Number of records limit of 1000 exceeded."
super().__init__("InvalidChangeBatch", message)
class NoSuchDelegationSet(Route53ClientError):
code = 400
def __init__(self, delegation_set_id):
super().__init__("NoSuchDelegationSet", delegation_set_id)
self.content_type = "text/xml"

79
moto/route53/models.py
View File

@ -11,6 +11,7 @@ from jinja2 import Template
from moto.route53.exceptions import (
InvalidInput,
NoSuchCloudWatchLogsLogGroup,
NoSuchDelegationSet,
NoSuchHostedZone,
NoSuchQueryLoggingConfig,
QueryLoggingConfigAlreadyExists,
@ -27,6 +28,21 @@ def create_route53_zone_id():
return "".join([random.choice(ROUTE53_ID_CHOICE) for _ in range(0, 15)])
class DelegationSet(BaseModel):
def __init__(self, caller_reference, name_servers, delegation_set_id):
self.caller_reference = caller_reference
self.name_servers = name_servers or [
"ns-2048.awsdns-64.com",
"ns-2049.awsdns-65.net",
"ns-2050.awsdns-66.org",
"ns-2051.awsdns-67.co.uk",
]
self.id = delegation_set_id or "".join(
[random.choice(ROUTE53_ID_CHOICE) for _ in range(5)]
)
self.location = f"https://route53.amazonaws.com/delegationset/{self.id}"
class HealthCheck(CloudFormationModel):
def __init__(self, health_check_id, caller_reference, health_check_args):
self.id = health_check_id
@ -213,7 +229,14 @@ def reverse_domain_name(domain_name):
class FakeZone(CloudFormationModel):
def __init__(
self, name, id_, private_zone, vpcid=None, vpcregion=None, comment=None
self,
name,
id_,
private_zone,
vpcid=None,
vpcregion=None,
comment=None,
delegation_set=None,
):
self.name = name
self.id = id_
@ -225,6 +248,7 @@ class FakeZone(CloudFormationModel):
self.vpcregion = vpcregion
self.private_zone = private_zone
self.rrsets = []
self.delegation_set = delegation_set
def add_rrset(self, record_set):
record_set = RecordSet(record_set)
@ -370,11 +394,21 @@ class Route53Backend(BaseBackend):
self.health_checks = {}
self.resource_tags = defaultdict(dict)
self.query_logging_configs = {}
self.delegation_sets = dict()
def create_hosted_zone(
self, name, private_zone, vpcid=None, vpcregion=None, comment=None
self,
name,
private_zone,
vpcid=None,
vpcregion=None,
comment=None,
delegation_set_id=None,
):
new_id = create_route53_zone_id()
delegation_set = self.create_reusable_delegation_set(
caller_reference=f"DelSet_{name}", delegation_set_id=delegation_set_id
)
new_zone = FakeZone(
name,
new_id,
@ -382,6 +416,7 @@ class Route53Backend(BaseBackend):
vpcid=vpcid,
vpcregion=vpcregion,
comment=comment,
delegation_set=delegation_set,
)
self.zones[new_id] = new_zone
return new_zone
@ -407,9 +442,18 @@ class Route53Backend(BaseBackend):
return self.resource_tags[resource_id]
return {}
def list_resource_record_sets(self, zone_id, start_type, start_name):
def list_resource_record_sets(self, zone_id, start_type, start_name, max_items):
"""
The StartRecordIdentifier-parameter is not yet implemented
"""
the_zone = self.get_hosted_zone(zone_id)
return the_zone.get_record_sets(start_type, start_name)
all_records = list(the_zone.get_record_sets(start_type, start_name))
records = all_records[0:max_items]
next_record = all_records[max_items] if len(all_records) > max_items else None
next_start_name = next_record.name if next_record else None
next_start_type = next_record.type_ if next_record else None
is_truncated = next_record is not None
return records, next_start_name, next_start_type, is_truncated
def change_resource_record_sets(self, zoneid, change_list):
the_zone = self.get_hosted_zone(zoneid)
@ -609,5 +653,32 @@ class Route53Backend(BaseBackend):
return list(self.query_logging_configs.values())
def create_reusable_delegation_set(
self, caller_reference, delegation_set_id=None, hosted_zone_id=None
):
name_servers = None
if hosted_zone_id:
hosted_zone = self.get_hosted_zone(hosted_zone_id)
name_servers = hosted_zone.delegation_set.name_servers
delegation_set = DelegationSet(
caller_reference, name_servers, delegation_set_id
)
self.delegation_sets[delegation_set.id] = delegation_set
return delegation_set
def list_reusable_delegation_sets(self):
"""
Pagination is not yet implemented
"""
return self.delegation_sets.values()
def delete_reusable_delegation_set(self, delegation_set_id):
self.delegation_sets.pop(delegation_set_id, None)
def get_reusable_delegation_set(self, delegation_set_id):
if delegation_set_id not in self.delegation_sets:
raise NoSuchDelegationSet(delegation_set_id)
return self.delegation_sets[delegation_set_id]
route53_backend = Route53Backend()

129
moto/route53/responses.py
View File

@ -56,6 +56,7 @@ class Route53(BaseResponse):
if name[-1] != ".":
name += "."
delegation_set_id = zone_request.get("DelegationSetId")
new_zone = route53_backend.create_hosted_zone(
name,
@ -63,6 +64,7 @@ class Route53(BaseResponse):
private_zone=private_zone,
vpcid=vpcid,
vpcregion=vpcregion,
delegation_set_id=delegation_set_id,
)
template = Template(CREATE_HOSTED_ZONE_RESPONSE)
return 201, headers, template.render(zone=new_zone)
@ -163,14 +165,30 @@ class Route53(BaseResponse):
template = Template(LIST_RRSET_RESPONSE)
start_type = querystring.get("type", [None])[0]
start_name = querystring.get("name", [None])[0]
max_items = int(querystring.get("maxitems", ["300"])[0])
if start_type and not start_name:
return 400, headers, "The input is not valid"
record_sets = route53_backend.list_resource_record_sets(
zoneid, start_type=start_type, start_name=start_name
(
record_sets,
next_name,
next_type,
is_truncated,
) = route53_backend.list_resource_record_sets(
zoneid,
start_type=start_type,
start_name=start_name,
max_items=max_items,
)
return 200, headers, template.render(record_sets=record_sets)
template = template.render(
record_sets=record_sets,
next_name=next_name,
next_type=next_type,
max_items=max_items,
is_truncated=is_truncated,
)
return 200, headers, template
def health_check_response(self, request, full_url, headers):
self.setup_class(request, full_url, headers)
@ -332,6 +350,52 @@ class Route53(BaseResponse):
route53_backend.delete_query_logging_config(query_logging_config_id)
return 200, headers, ""
def reusable_delegation_sets(self, request, full_url, headers):
self.setup_class(request, full_url, headers)
if request.method == "GET":
delegation_sets = route53_backend.list_reusable_delegation_sets()
template = self.response_template(LIST_REUSABLE_DELEGATION_SETS_TEMPLATE)
return (
200,
{},
template.render(
delegation_sets=delegation_sets,
marker=None,
is_truncated=False,
max_items=100,
),
)
elif request.method == "POST":
elements = xmltodict.parse(self.body)
root_elem = elements["CreateReusableDelegationSetRequest"]
caller_reference = root_elem.get("CallerReference")
hosted_zone_id = root_elem.get("HostedZoneId")
delegation_set = route53_backend.create_reusable_delegation_set(
caller_reference=caller_reference, hosted_zone_id=hosted_zone_id,
)
template = self.response_template(CREATE_REUSABLE_DELEGATION_SET_TEMPLATE)
return (
201,
{"Location": delegation_set.location},
template.render(delegation_set=delegation_set),
)
@error_handler
def reusable_delegation_set(self, request, full_url, headers):
self.setup_class(request, full_url, headers)
parsed_url = urlparse(full_url)
ds_id = parsed_url.path.rstrip("/").rsplit("/")[-1]
if request.method == "GET":
delegation_set = route53_backend.get_reusable_delegation_set(
delegation_set_id=ds_id
)
template = self.response_template(GET_REUSABLE_DELEGATION_SET_TEMPLATE)
return 200, {}, template.render(delegation_set=delegation_set)
if request.method == "DELETE":
route53_backend.delete_reusable_delegation_set(delegation_set_id=ds_id)
template = self.response_template(DELETE_REUSABLE_DELEGATION_SET_TEMPLATE)
return 200, {}, template.render()
LIST_TAGS_FOR_RESOURCE_RESPONSE = """
<ListTagsForResourceResponse xmlns="https://route53.amazonaws.com/doc/2015-01-01/">
@ -404,7 +468,10 @@ LIST_RRSET_RESPONSE = """<?xml version="1.0" encoding="UTF-8"?>
</ResourceRecordSet>
{% endfor %}
</ResourceRecordSets>
<IsTruncated>false</IsTruncated>
{% if is_truncated %}<NextRecordName>{{ next_name }}</NextRecordName>{% endif %}
{% if is_truncated %}<NextRecordType>{{ next_type }}</NextRecordType>{% endif %}
<MaxItems>{{ max_items }}</MaxItems>
<IsTruncated>{{ 'true' if is_truncated else 'false' }}</IsTruncated>
</ListResourceRecordSetsResponse>"""
CHANGE_RRSET_RESPONSE = """<ChangeResourceRecordSetsResponse xmlns="https://route53.amazonaws.com/doc/2012-12-12/">
@ -438,8 +505,9 @@ GET_HOSTED_ZONE_RESPONSE = """<GetHostedZoneResponse xmlns="https://route53.amaz
</Config>
</HostedZone>
<DelegationSet>
<Id>{{ zone.delegation_set.id }}</Id>
<NameServers>
<NameServer>moto.test.com</NameServer>
{% for name in zone.delegation_set.name_servers %}<NameServer>{{ name }}</NameServer>{% endfor %}
</NameServers>
</DelegationSet>
<VPCs>
@ -464,8 +532,9 @@ CREATE_HOSTED_ZONE_RESPONSE = """<CreateHostedZoneResponse xmlns="https://route5
</Config>
</HostedZone>
<DelegationSet>
<Id>{{ zone.delegation_set.id }}</Id>
<NameServers>
<NameServer>moto.test.com</NameServer>
{% for name in zone.delegation_set.name_servers %}<NameServer>{{ name }}</NameServer>{% endfor %}
</NameServers>
</DelegationSet>
<VPC>
@ -584,3 +653,51 @@ LIST_QUERY_LOGGING_CONFIGS_RESPONSE = """<?xml version="1.0" encoding="UTF-8"?>
<NextToken>{{ next_token }}</NextToken>
{% endif %}
</ListQueryLoggingConfigsResponse>"""
CREATE_REUSABLE_DELEGATION_SET_TEMPLATE = """<CreateReusableDelegationSetResponse>
<DelegationSet>
<Id>{{ delegation_set.id }}</Id>
<CallerReference>{{ delegation_set.caller_reference }}</CallerReference>
<NameServers>
{% for name in delegation_set.name_servers %}<NameServer>{{ name }}</NameServer>{% endfor %}
</NameServers>
</DelegationSet>
</CreateReusableDelegationSetResponse>
"""
LIST_REUSABLE_DELEGATION_SETS_TEMPLATE = """<ListReusableDelegationSetsResponse>
<DelegationSets>
{% for delegation in delegation_sets %}
<DelegationSet>
<Id>{{ delegation.id }}</Id>
<CallerReference>{{ delegation.caller_reference }}</CallerReference>
<NameServers>
{% for name in delegation.name_servers %}<NameServer>{{ name }}</NameServer>{% endfor %}
</NameServers>
</DelegationSet>
{% endfor %}
</DelegationSets>
<Marker>{{ marker }}</Marker>
<IsTruncated>{{ is_truncated }}</IsTruncated>
<MaxItems>{{ max_items }}</MaxItems>
</ListReusableDelegationSetsResponse>
"""
DELETE_REUSABLE_DELEGATION_SET_TEMPLATE = """<DeleteReusableDelegationSetResponse>
<DeleteReusableDelegationSetResponse/>
</DeleteReusableDelegationSetResponse>
"""
GET_REUSABLE_DELEGATION_SET_TEMPLATE = """<GetReusableDelegationSetResponse>
<DelegationSet>
<Id>{{ delegation_set.id }}</Id>
<CallerReference>{{ delegation_set.caller_reference }}</CallerReference>
<NameServers>
{% for name in delegation_set.name_servers %}<NameServer>{{ name }}</NameServer>{% endfor %}
</NameServers>
</DelegationSet>
</GetReusableDelegationSetResponse>
"""

2
moto/route53/urls.py
View File

@ -27,4 +27,6 @@ url_paths = {
r"{0}/(?P<api_version>[\d_-]+)/change/(?P<change_id>[^/]+)$": Route53().get_change,
r"{0}/(?P<api_version>[\d_-]+)/queryloggingconfig$": Route53().list_or_create_query_logging_config_response,
r"{0}/(?P<api_version>[\d_-]+)/queryloggingconfig/(?P<query_id>[^/]+)$": Route53().get_or_delete_query_logging_config_response,
r"{0}/(?P<api_version>[\d_-]+)/delegationset$": Route53().reusable_delegation_sets,
r"{0}/(?P<api_version>[\d_-]+)/delegationset/(?P<delegation_set_id>[^/]+)$": Route53().reusable_delegation_set,
}

72
tests/test_route53/test_route53.py
View File

@ -22,7 +22,11 @@ def test_create_hosted_zone_boto3():
firstzone.should.have.key("ResourceRecordSetCount").equal(0)
delegation = response["DelegationSet"]
delegation.should.equal({"NameServers": ["moto.test.com"]})
delegation.should.have.key("NameServers").length_of(4)
delegation["NameServers"].should.contain("ns-2048.awsdns-64.com")
delegation["NameServers"].should.contain("ns-2049.awsdns-65.net")
delegation["NameServers"].should.contain("ns-2050.awsdns-66.org")
delegation["NameServers"].should.contain("ns-2051.awsdns-67.co.uk")
@mock_route53
@ -1419,3 +1423,69 @@ def test_change_resource_record_sets_records_limit():
err = exc.value.response["Error"]
err["Code"].should.equal("InvalidChangeBatch")
err["Message"].should.equal("Number of records limit of 1000 exceeded.")
@mock_route53
def test_list_resource_recordset_pagination():
conn = boto3.client("route53", region_name="us-east-1")
conn.create_hosted_zone(
Name="db.",
CallerReference=str(hash("foo")),
HostedZoneConfig=dict(PrivateZone=True, Comment="db"),
)
zones = conn.list_hosted_zones_by_name(DNSName="db.")
len(zones["HostedZones"]).should.equal(1)
zones["HostedZones"][0]["Name"].should.equal("db.")
hosted_zone_id = zones["HostedZones"][0]["Id"]
# Create A Record.
a_record_endpoint_payload = {
"Comment": f"Create 500 A records",
"Changes": [
{
"Action": "CREATE",
"ResourceRecordSet": {
"Name": f"env{idx}.redis.db.",
"Type": "A",
"TTL": 10,
"ResourceRecords": [{"Value": "127.0.0.1"}],
},
}
for idx in range(500)
],
}
conn.change_resource_record_sets(
HostedZoneId=hosted_zone_id, ChangeBatch=a_record_endpoint_payload
)
response = conn.list_resource_record_sets(
HostedZoneId=hosted_zone_id, MaxItems="100"
)
response.should.have.key("ResourceRecordSets").length_of(100)
response.should.have.key("IsTruncated").equals(True)
response.should.have.key("MaxItems").equals("100")
response.should.have.key("NextRecordName").equals("env189.redis.db.")
response.should.have.key("NextRecordType").equals("A")
response = conn.list_resource_record_sets(
HostedZoneId=hosted_zone_id,
StartRecordName=response["NextRecordName"],
StartRecordType=response["NextRecordType"],
)
response.should.have.key("ResourceRecordSets").length_of(300)
response.should.have.key("IsTruncated").equals(True)
response.should.have.key("MaxItems").equals("300")
response.should.have.key("NextRecordName").equals("env459.redis.db.")
response.should.have.key("NextRecordType").equals("A")
response = conn.list_resource_record_sets(
HostedZoneId=hosted_zone_id,
StartRecordName=response["NextRecordName"],
StartRecordType=response["NextRecordType"],
)
response.should.have.key("ResourceRecordSets").length_of(100)
response.should.have.key("IsTruncated").equals(False)
response.should.have.key("MaxItems").equals("300")
response.shouldnt.have.key("NextRecordName")
response.shouldnt.have.key("NextRecordType")

120
tests/test_route53/test_route53_delegationsets.py Normal file
View File

@ -0,0 +1,120 @@
import boto3
import pytest
from botocore.exceptions import ClientError
from moto import mock_route53
@mock_route53
def test_list_reusable_delegation_set():
client = boto3.client("route53", region_name="us-east-1")
resp = client.list_reusable_delegation_sets()
resp.should.have.key("DelegationSets").equals([])
resp.should.have.key("IsTruncated").equals(False)
@mock_route53
def test_create_reusable_delegation_set():
client = boto3.client("route53", region_name="us-east-1")
resp = client.create_reusable_delegation_set(CallerReference="r3f3r3nc3")
headers = resp["ResponseMetadata"]["HTTPHeaders"]
headers.should.have.key("location")
resp.should.have.key("DelegationSet")
resp["DelegationSet"].should.have.key("Id")
resp["DelegationSet"].should.have.key("CallerReference").equals("r3f3r3nc3")
resp["DelegationSet"].should.have.key("NameServers").length_of(4)
@mock_route53
def test_create_reusable_delegation_set_from_hosted_zone():
client = boto3.client("route53", region_name="us-east-1")
response = client.create_hosted_zone(
Name="testdns.aws.com.", CallerReference=str(hash("foo"))
)
hosted_zone_id = response["HostedZone"]["Id"]
print(response)
hosted_zone_name_servers = set(response["DelegationSet"]["NameServers"])
resp = client.create_reusable_delegation_set(
CallerReference="r3f3r3nc3", HostedZoneId=hosted_zone_id
)
set(resp["DelegationSet"]["NameServers"]).should.equal(hosted_zone_name_servers)
@mock_route53
def test_create_reusable_delegation_set_from_hosted_zone_with_delegationsetid():
client = boto3.client("route53", region_name="us-east-1")
response = client.create_hosted_zone(
Name="testdns.aws.com.",
CallerReference=str(hash("foo")),
DelegationSetId="customdelegationsetid",
)
response.should.have.key("DelegationSet")
response["DelegationSet"].should.have.key("Id").equals("customdelegationsetid")
response["DelegationSet"].should.have.key("NameServers")
hosted_zone_id = response["HostedZone"]["Id"]
hosted_zone_name_servers = set(response["DelegationSet"]["NameServers"])
resp = client.create_reusable_delegation_set(
CallerReference="r3f3r3nc3", HostedZoneId=hosted_zone_id
)
resp["DelegationSet"].should.have.key("Id").shouldnt.equal("customdelegationsetid")
set(resp["DelegationSet"]["NameServers"]).should.equal(hosted_zone_name_servers)
@mock_route53
def test_get_reusable_delegation_set():
client = boto3.client("route53", region_name="us-east-1")
ds_id = client.create_reusable_delegation_set(CallerReference="r3f3r3nc3")[
"DelegationSet"
]["Id"]
resp = client.get_reusable_delegation_set(Id=ds_id)
resp.should.have.key("DelegationSet")
resp["DelegationSet"].should.have.key("Id").equals(ds_id)
resp["DelegationSet"].should.have.key("CallerReference").equals("r3f3r3nc3")
resp["DelegationSet"].should.have.key("NameServers").length_of(4)
@mock_route53
def test_get_reusable_delegation_set_unknown():
client = boto3.client("route53", region_name="us-east-1")
with pytest.raises(ClientError) as exc:
client.get_reusable_delegation_set(Id="unknown")
err = exc.value.response["Error"]
err["Code"].should.equal("NoSuchDelegationSet")
err["Message"].should.equal("unknown")
@mock_route53
def test_list_reusable_delegation_sets():
client = boto3.client("route53", region_name="us-east-1")
client.create_reusable_delegation_set(CallerReference="r3f3r3nc3")
client.create_reusable_delegation_set(CallerReference="r3f3r3nc4")
resp = client.list_reusable_delegation_sets()
resp.should.have.key("DelegationSets").length_of(2)
resp.should.have.key("IsTruncated").equals(False)
@mock_route53
def test_delete_reusable_delegation_set():
client = boto3.client("route53", region_name="us-east-1")
ds_id = client.create_reusable_delegation_set(CallerReference="r3f3r3nc3")[
"DelegationSet"
]["Id"]
client.delete_reusable_delegation_set(Id=ds_id)
client.list_reusable_delegation_sets()["DelegationSets"].should.have.length_of(0)