From d815421072bbeec6c15129bb9e1b3a0fb3d67236 Mon Sep 17 00:00:00 2001 From: Neev Cohen <70970900+NeevCohen@users.noreply.github.com> Date: Tue, 5 Mar 2024 01:16:02 +0200 Subject: [PATCH] Feature: Route53Domains (#7406) --- moto/backend_index.py | 4 + moto/backends.py | 4 + moto/core/responses.py | 4 +- moto/route53domains/__init__.py | 1 + moto/route53domains/exceptions.py | 43 + moto/route53domains/models.py | 302 +++++ moto/route53domains/responses.py | 138 ++ moto/route53domains/urls.py | 11 + moto/route53domains/validators.py | 1189 +++++++++++++++++ tests/test_route53domains/__init__.py | 0 .../test_route53domains_domain.py | 562 ++++++++ 11 files changed, 2255 insertions(+), 3 deletions(-) create mode 100644 moto/route53domains/__init__.py create mode 100644 moto/route53domains/exceptions.py create mode 100644 moto/route53domains/models.py create mode 100644 moto/route53domains/responses.py create mode 100644 moto/route53domains/urls.py create mode 100644 moto/route53domains/validators.py create mode 100644 tests/test_route53domains/__init__.py create mode 100644 tests/test_route53domains/test_route53domains_domain.py diff --git a/moto/backend_index.py b/moto/backend_index.py index c413f9288..5e0d4455b 100644 --- a/moto/backend_index.py +++ b/moto/backend_index.py @@ -142,6 +142,10 @@ backend_url_patterns = [ "route53resolver", re.compile("https?://route53resolver\\.(.+)\\.amazonaws\\.com"), ), + ( + "route53domains", + re.compile("https?://route53domains\\.(.+)\\.amazonaws\\.com"), + ), ("s3", re.compile("https?://s3(?!-control)(.*)\\.amazonaws.com")), ( "s3", diff --git a/moto/backends.py b/moto/backends.py index 14ba4ad92..e7d636f1e 100644 --- a/moto/backends.py +++ b/moto/backends.py @@ -108,6 +108,7 @@ if TYPE_CHECKING: from moto.resourcegroupstaggingapi.models import ResourceGroupsTaggingAPIBackend from moto.robomaker.models import RoboMakerBackend from moto.route53.models import Route53Backend + from moto.route53domains.models import Route53DomainsBackend from moto.route53resolver.models import Route53ResolverBackend from moto.s3.models import S3Backend from moto.s3control.models import S3ControlBackend @@ -266,6 +267,7 @@ SERVICE_NAMES = Union[ "Literal['robomaker']", "Literal['route53']", "Literal['route53resolver']", + "Literal['route53domains']", "Literal['s3']", "Literal['s3bucket_path']", "Literal['s3control']", @@ -506,6 +508,8 @@ def get_backend(name: "Literal['route53']") -> "BackendDict[Route53Backend]": .. @overload def get_backend(name: "Literal['route53resolver']") -> "BackendDict[Route53ResolverBackend]": ... @overload +def get_backend(name: "Literal['route53domains']") -> "BackendDict[Route53DomainsBackend]": ... +@overload def get_backend(name: "Literal['s3']") -> "BackendDict[S3Backend]": ... @overload def get_backend(name: "Literal['s3bucket_path']") -> "BackendDict[S3Backend]": ... diff --git a/moto/core/responses.py b/moto/core/responses.py index b6ae0270e..d091b2157 100644 --- a/moto/core/responses.py +++ b/moto/core/responses.py @@ -633,9 +633,7 @@ class BaseResponse(_TemplateEnvironmentMixin, ActionAuthenticatorMixin): if self.body is not None: try: return json.loads(self.body)[param_name] - except ValueError: - pass - except KeyError: + except (ValueError, KeyError): pass # try to get path parameter if self.uri_match: diff --git a/moto/route53domains/__init__.py b/moto/route53domains/__init__.py new file mode 100644 index 000000000..132e14902 --- /dev/null +++ b/moto/route53domains/__init__.py @@ -0,0 +1 @@ +from .models import route53domains_backends # noqa: F401 diff --git a/moto/route53domains/exceptions.py b/moto/route53domains/exceptions.py new file mode 100644 index 000000000..bb4714ce6 --- /dev/null +++ b/moto/route53domains/exceptions.py @@ -0,0 +1,43 @@ +from typing import List + +from moto.core.exceptions import JsonRESTError + + +class DomainLimitExceededException(JsonRESTError): + code = 400 + + def __init__(self) -> None: + super().__init__( + "DomainLimitExceeded", + "The number of registered domains has exceeded the allowed threshold for this account. If you want to " + "register more domains please request a higher quota", + ) + + +class DuplicateRequestException(JsonRESTError): + code = 400 + + def __init__(self) -> None: + super().__init__( + "DuplicateRequest", "The request is already in progress for the domain." + ) + + +class InvalidInputException(JsonRESTError): + code = 400 + + def __init__(self, error_msgs: List[str]): + error_msgs_str = "\n\t".join(error_msgs) + super().__init__( + "InvalidInput", f"The requested item is not acceptable.\n\t{error_msgs_str}" + ) + + +class UnsupportedTLDException(JsonRESTError): + code = 400 + + def __init__(self, tld: str): + super().__init__( + "UnsupportedTLD", + f"Amazon Route53 does not support the top-level domain (TLD) `.{tld}`.", + ) diff --git a/moto/route53domains/models.py b/moto/route53domains/models.py new file mode 100644 index 000000000..3208b6f8d --- /dev/null +++ b/moto/route53domains/models.py @@ -0,0 +1,302 @@ +from datetime import datetime, timedelta, timezone +from typing import Any, Dict, List, Optional + +from moto.core.base_backend import BackendDict, BaseBackend +from moto.route53 import route53_backends +from moto.route53.models import Route53Backend +from moto.utilities.paginator import paginate + +from .exceptions import ( + DomainLimitExceededException, + DuplicateRequestException, + InvalidInputException, +) +from .validators import ( + DOMAIN_OPERATION_STATUSES, + DOMAIN_OPERATION_TYPES, + DomainFilterField, + DomainsFilter, + DomainSortOrder, + DomainsSortCondition, + NameServer, + Route53Domain, + Route53DomainsContactDetail, + Route53DomainsOperation, + ValidationException, +) + + +class Route53DomainsBackend(BaseBackend): + """Implementation of Route53Domains APIs.""" + + DEFAULT_MAX_DOMAINS_COUNT = 20 + PAGINATION_MODEL = { + "list_domains": { + "input_token": "marker", + "limit_key": "max_items", + "limit_default": 20, + "unique_attribute": "domain_name", + }, + "list_operations": { + "input_token": "marker", + "limit_key": "max_items", + "limit_default": 20, + "unique_attribute": "id", + }, + } + + def __init__(self, region_name: str, account_id: str): + super().__init__(region_name, account_id) + self.__route53_backend: Route53Backend = route53_backends[account_id]["global"] + self.__domains: Dict[str, Route53Domain] = {} + self.__operations: Dict[str, Route53DomainsOperation] = {} + + def register_domain( + self, + domain_name: str, + duration_in_years: int, + auto_renew: bool, + admin_contact: Dict[str, Any], + registrant_contact: Dict[str, Any], + tech_contact: Dict[str, Any], + private_protect_admin_contact: bool, + private_protect_registrant_contact: bool, + private_protect_tech_contact: bool, + extra_params: List[Dict[str, Any]], + ) -> Route53DomainsOperation: + """Register a domain""" + + if len(self.__domains) == self.DEFAULT_MAX_DOMAINS_COUNT: + raise DomainLimitExceededException() + + requested_operation = Route53DomainsOperation.validate( + domain_name=domain_name, status="SUCCESSFUL", type_="REGISTER_DOMAIN" + ) + + self.__validate_duplicate_operations(requested_operation) + + expiration_date = datetime.now(timezone.utc) + timedelta( + days=365 * duration_in_years + ) + + try: + + domain = Route53Domain.validate( + domain_name=domain_name, + auto_renew=auto_renew, + admin_contact=Route53DomainsContactDetail.validate_dict(admin_contact), + registrant_contact=Route53DomainsContactDetail.validate_dict( + registrant_contact + ), + tech_contact=Route53DomainsContactDetail.validate_dict(tech_contact), + admin_privacy=private_protect_admin_contact, + registrant_privacy=private_protect_registrant_contact, + tech_privacy=private_protect_tech_contact, + expiration_date=expiration_date, + extra_params=extra_params, + ) + + except ValidationException as e: + raise InvalidInputException(e.errors) + self.__operations[requested_operation.id] = requested_operation + + self.__route53_backend.create_hosted_zone( + name=domain.domain_name, private_zone=False + ) + + self.__domains[domain_name] = domain + return requested_operation + + def delete_domain(self, domain_name: str) -> Route53DomainsOperation: + requested_operation = Route53DomainsOperation.validate( + domain_name=domain_name, status="SUCCESSFUL", type_="DELETE_DOMAIN" + ) + self.__validate_duplicate_operations(requested_operation) + + input_errors: List[str] = [] + Route53Domain.validate_domain_name(domain_name, input_errors) + + if input_errors: + raise InvalidInputException(input_errors) + + if domain_name not in self.__domains: + raise InvalidInputException( + [f"Domain {domain_name} isn't registered in the current account"] + ) + + self.__operations[requested_operation.id] = requested_operation + del self.__domains[domain_name] + return requested_operation + + def __validate_duplicate_operations( + self, requested_operation: Route53DomainsOperation + ) -> None: + for operation in self.__operations.values(): + if ( + operation.domain_name == requested_operation.domain_name + and operation.type == requested_operation.type + ): + raise DuplicateRequestException() + + def get_domain(self, domain_name: str) -> Route53Domain: + input_errors: List[str] = [] + Route53Domain.validate_domain_name(domain_name, input_errors) + if input_errors: + raise InvalidInputException(input_errors) + + if domain_name not in self.__domains: + raise InvalidInputException(["Domain is not associated with this account"]) + + return self.__domains[domain_name] + + @paginate(pagination_model=PAGINATION_MODEL) # type: ignore[misc] + def list_domains( + self, + filter_conditions: Optional[List[Dict[str, Any]]] = None, + sort_condition: Optional[Dict[str, Any]] = None, + ) -> List[Route53Domain]: + try: + filters: List[DomainsFilter] = ( + [DomainsFilter.validate_dict(f) for f in filter_conditions] + if filter_conditions + else [] + ) + sort: Optional[DomainsSortCondition] = ( + DomainsSortCondition.validate_dict(sort_condition) + if sort_condition + else None + ) + except ValidationException as e: + raise InvalidInputException(e.errors) + + filter_fields = [f.name for f in filters] + if sort and filter_fields and sort.name not in filter_fields: + raise InvalidInputException( + ["Sort condition must be the same as the filter condition"] + ) + + domains_to_return: List[Route53Domain] = [] + + for domain in self.__domains.values(): + if all([f.filter(domain) for f in filters]): + domains_to_return.append(domain) + + if sort: + if sort.name == DomainFilterField.DOMAIN_NAME: + domains_to_return.sort( + key=lambda d: d.domain_name, + reverse=(sort.sort_order == DomainSortOrder.DESCENDING), + ) + else: + domains_to_return.sort( + key=lambda d: d.expiration_date, + reverse=(sort.sort_order == DomainSortOrder.DESCENDING), + ) + + return domains_to_return + + @paginate(pagination_model=PAGINATION_MODEL) # type: ignore[misc] + def list_operations( + self, + submitted_since_timestamp: Optional[int] = None, + statuses: Optional[List[str]] = None, + types: Optional[List[str]] = None, + sort_by: Optional[str] = None, + sort_order: Optional[str] = None, + ) -> List[Route53DomainsOperation]: + + input_errors: List[str] = [] + statuses = statuses or [] + types = types or [] + + if any(status not in DOMAIN_OPERATION_STATUSES for status in statuses): + input_errors.append("Status is invalid") + if any(type_ not in DOMAIN_OPERATION_TYPES for type_ in types): + input_errors.append("Type is invalid") + + if input_errors: + raise InvalidInputException(input_errors) + + submitted_since = ( + datetime.fromtimestamp(submitted_since_timestamp, timezone.utc) + if submitted_since_timestamp + else None + ) + + operations_to_return: List[Route53DomainsOperation] = [] + + for operation in self.__operations.values(): + if statuses and operation.status not in statuses: + continue + + if types and operation.type not in types: + continue + + if submitted_since and operation.submitted_date < submitted_since: + continue + + operations_to_return.append(operation) + + if sort_by == "SubmittedDate": + operations_to_return.sort( + key=lambda op: op.submitted_date, + reverse=sort_order == DomainSortOrder.ASCENDING, + ) + + return operations_to_return + + def get_operation(self, operation_id: str) -> Route53DomainsOperation: + if operation_id not in self.__operations: + raise InvalidInputException( + [f"Operation with id {operation_id} doesn't exist"] + ) + + return self.__operations[operation_id] + + def update_domain_nameservers( + self, domain_name: str, nameservers: List[Dict[str, Any]] + ) -> Route53DomainsOperation: + input_errors: List[str] = [] + Route53Domain.validate_domain_name(domain_name, input_errors) + if len(nameservers) < 1: + input_errors.append("Must supply nameservers") + + servers: List[NameServer] = [] + try: + servers = [NameServer.validate_dict(obj) for obj in nameservers] + except ValidationException as e: + input_errors += e.errors + + for server in servers: + if domain_name in server.name and not server.glue_ips: + input_errors.append( + f"Must supply glue IPs for name server {server.name} because it is a subdomain of " + f"the domain" + ) + + if input_errors: + raise InvalidInputException(input_errors) + + if domain_name not in self.__domains: + raise InvalidInputException( + [f"Domain {domain_name} is not registered to the current AWS account"] + ) + + requested_operation = Route53DomainsOperation.validate( + domain_name=domain_name, status="SUCCESSFUL", type_="UPDATE_NAMESERVER" + ) + self.__validate_duplicate_operations(requested_operation) + + domain = self.__domains[domain_name] + domain.nameservers = servers + self.__operations[requested_operation.id] = requested_operation + + return requested_operation + + +route53domains_backends = BackendDict( + Route53DomainsBackend, + "route53domains", + use_boto3_regions=False, + additional_regions=["global"], +) diff --git a/moto/route53domains/responses.py b/moto/route53domains/responses.py new file mode 100644 index 000000000..78287c7bc --- /dev/null +++ b/moto/route53domains/responses.py @@ -0,0 +1,138 @@ +import json +from typing import Any, Dict + +from moto.core.responses import BaseResponse +from moto.route53domains.models import Route53DomainsBackend, route53domains_backends +from moto.route53domains.validators import Route53Domain, Route53DomainsOperation + + +class Route53DomainsResponse(BaseResponse): + def __init__(self) -> None: + super().__init__(service_name="route53-domains") + + @property + def route53domains_backend(self) -> Route53DomainsBackend: + return route53domains_backends[self.current_account]["global"] + + def register_domain(self) -> str: + domain_name = self._get_param("DomainName") + duration_in_years = self._get_int_param("DurationInYears") + auto_renew = self._get_bool_param("AutoRenew", if_none=True) + admin_contact = self._get_param("AdminContact") + registrant_contact = self._get_param("RegistrantContact") + tech_contact = self._get_param("TechContact") + privacy_protection_admin_contact = self._get_bool_param( + "PrivacyProtectAdminContact", if_none=True + ) + privacy_protection_registrant_contact = self._get_bool_param( + "PrivacyProtectRegistrantContact", if_none=True + ) + privacy_protection_tech_contact = self._get_bool_param( + "PrivacyProtectTechContact", if_none=True + ) + extra_params = self._get_param("ExtraParams") + + operation = self.route53domains_backend.register_domain( + domain_name=domain_name, + duration_in_years=duration_in_years, + auto_renew=auto_renew, + admin_contact=admin_contact, + registrant_contact=registrant_contact, + tech_contact=tech_contact, + private_protect_admin_contact=privacy_protection_admin_contact, + private_protect_registrant_contact=privacy_protection_registrant_contact, + private_protect_tech_contact=privacy_protection_tech_contact, + extra_params=extra_params, + ) + + return json.dumps({"OperationId": operation.id}) + + def delete_domain(self) -> str: + domain_name = self._get_param("DomainName") + operation = self.route53domains_backend.delete_domain(domain_name) + + return json.dumps({"OperationId": operation.id}) + + def get_domain_detail(self) -> str: + domain_name = self._get_param("DomainName") + + return json.dumps( + self.route53domains_backend.get_domain(domain_name=domain_name).to_json() + ) + + def list_domains(self) -> str: + filter_conditions = self._get_param("FilterConditions") + sort_condition = self._get_param("SortCondition") + marker = self._get_param("Marker") + max_items = self._get_param("MaxItems") + domains, marker = self.route53domains_backend.list_domains( + filter_conditions=filter_conditions, + sort_condition=sort_condition, + marker=marker, + max_items=max_items, + ) + res = { + "Domains": list(map(self.__map_domains_to_info, domains)), + } + + if marker: + res["NextPageMarker"] = marker + + return json.dumps(res) + + @staticmethod + def __map_domains_to_info(domain: Route53Domain) -> Dict[str, Any]: # type: ignore[misc] + return { + "DomainName": domain.domain_name, + "AutoRenew": domain.auto_renew, + "Expiry": domain.expiration_date.timestamp(), + "TransferLock": True, + } + + def list_operations(self) -> str: + submitted_since_timestamp = self._get_int_param("SubmittedSince") + max_items = self._get_int_param("MaxItems") + statuses = self._get_param("Status") + marker = self._get_param("Marker") + types = self._get_param("Type") + sort_by = self._get_param("SortBy") + sort_order = self._get_param("SortOrder") + + operations, marker = self.route53domains_backend.list_operations( + submitted_since_timestamp=submitted_since_timestamp, + max_items=max_items, + marker=marker, + statuses=statuses, + types=types, + sort_by=sort_by, + sort_order=sort_order, + ) + + res = { + "Operations": [operation.to_json() for operation in operations], + } + + if marker: + res["NextPageMarker"] = marker + + return json.dumps(res) + + def get_operation_detail(self) -> str: + operation_id = self._get_param("OperationId") + operation: Route53DomainsOperation = self.route53domains_backend.get_operation( + operation_id + ) + + return json.dumps(operation.to_json()) + + def update_domain_nameservers(self) -> str: + domain_name = self._get_param("DomainName") + nameservers = self._get_param("Nameservers") + + operation: Route53DomainsOperation = ( + self.route53domains_backend.update_domain_nameservers( + domain_name=domain_name, nameservers=nameservers + ) + ) + + return json.dumps({"OperationId": operation.id}) diff --git a/moto/route53domains/urls.py b/moto/route53domains/urls.py new file mode 100644 index 000000000..500444d22 --- /dev/null +++ b/moto/route53domains/urls.py @@ -0,0 +1,11 @@ +"""route53domains base URL and path.""" +from .responses import Route53DomainsResponse + +url_bases = [ + r"https?://route53domains\.(.+)\.amazonaws\.com", +] + + +url_paths = { + "{0}/$": Route53DomainsResponse.dispatch, +} diff --git a/moto/route53domains/validators.py b/moto/route53domains/validators.py new file mode 100644 index 000000000..b5bfb0eb2 --- /dev/null +++ b/moto/route53domains/validators.py @@ -0,0 +1,1189 @@ +import re +from datetime import datetime, timedelta, timezone +from enum import Enum +from ipaddress import IPv4Address, IPv6Address, ip_address +from typing import Any, Dict, List, Optional, Type + +from moto.core.common_models import BaseModel +from moto.moto_api._internal import mock_random +from moto.route53domains.exceptions import UnsupportedTLDException + +DOMAIN_OPERATION_STATUSES = ( + "SUBMITTED", + "IN_PROGRESS", + "ERROR", + "SUCCESSFUL", + "FAILED", +) + +DOMAIN_OPERATION_TYPES = ( + "REGISTER_DOMAIN", + "DELETE_DOMAIN", + "TRANSFER_IN_DOMAIN", + "UPDATE_DOMAIN_CONTACT", + "UPDATE_NAMESERVER", + "CHANGE_PRIVACY_PROTECTION", + "DOMAIN_LOCK", + "ENABLE_AUTORENEW", + "DISABLE_AUTORENEW", + "ADD_DNSSEC", + "REMOVE_DNSSEC", + "EXPIRE_DOMAIN", + "TRANSFER_OUT_DOMAIN", + "CHANGE_DOMAIN_OWNER", + "RENEW_DOMAIN", + "PUSH_DOMAIN", + "INTERNAL_TRANSFER_OUT_DOMAIN", + "INTERNAL_TRANSFER_IN_DOMAIN", +) + +DOMAIN_OPERATION_STATUS_FLAGS = ( + "PENDING_ACCEPTANCE", + "PENDING_CUSTOMER_ACTION", + "PENDING_AUTHORIZATION", + "PENDING_PAYMENT_VERIFICATION", + "PENDING_SUPPORT_CASE", +) + +DOMAIN_CONTACT_DETAIL_CONTACT_TYPES = ( + "PERSON", + "COMPANY", + "ASSOCIATION", + "PUBLIC_BODY", + "RESELLER", + "ORGANIZATION", +) + +DOMAIN_CONTACT_DETAIL_COUNTRY_CODES = ( + "AC", + "AD", + "AE", + "AF", + "AG", + "AI", + "AL", + "AM", + "AN", + "AO", + "AQ", + "AR", + "AS", + "AT", + "AU", + "AW", + "AX", + "AZ", + "BA", + "BB", + "BD", + "BE", + "BF", + "BG", + "BH", + "BI", + "BJ", + "BL", + "BM", + "BN", + "BO", + "BQ", + "BR", + "BS", + "BT", + "BV", + "BW", + "BY", + "BZ", + "CA", + "CC", + "CD", + "CF", + "CG", + "CH", + "CI", + "CK", + "CL", + "CM", + "CN", + "CO", + "CR", + "CU", + "CV", + "CW", + "CX", + "CY", + "CZ", + "DE", + "DJ", + "DK", + "DM", + "DO", + "DZ", + "EC", + "EE", + "EG", + "EH", + "ER", + "ES", + "ET", + "FI", + "FJ", + "FK", + "FM", + "FO", + "FR", + "GA", + "GB", + "GD", + "GE", + "GF", + "GG", + "GH", + "GI", + "GL", + "GM", + "GN", + "GP", + "GQ", + "GR", + "GS", + "GT", + "GU", + "GW", + "GY", + "HK", + "HM", + "HN", + "HR", + "HT", + "HU", + "ID", + "IE", + "IL", + "IM", + "IN", + "IO", + "IQ", + "IR", + "IS", + "IT", + "JE", + "JM", + "JO", + "JP", + "KE", + "KG", + "KH", + "KI", + "KM", + "KN", + "KP", + "KR", + "KW", + "KY", + "KZ", + "LA", + "LB", + "LC", + "LI", + "LK", + "LR", + "LS", + "LT", + "LU", + "LV", + "LY", + "MA", + "MC", + "MD", + "ME", + "MF", + "MG", + "MH", + "MK", + "ML", + "MM", + "MN", + "MO", + "MP", + "MQ", + "MR", + "MS", + "MT", + "MU", + "MV", + "MW", + "MX", + "MY", + "MZ", + "NA", + "NC", + "NE", + "NF", + "NG", + "NI", + "NL", + "NO", + "NP", + "NR", + "NU", + "NZ", + "OM", + "PA", + "PE", + "PF", + "PG", + "PH", + "PK", + "PL", + "PM", + "PN", + "PR", + "PS", + "PT", + "PW", + "PY", + "QA", + "RE", + "RO", + "RS", + "RU", + "RW", + "SA", + "SB", + "SC", + "SD", + "SE", + "SG", + "SH", + "SI", + "SJ", + "SK", + "SL", + "SM", + "SN", + "SO", + "SR", + "SS", + "ST", + "SV", + "SX", + "SY", + "SZ", + "TC", + "TD", + "TF", + "TG", + "TH", + "TJ", + "TK", + "TL", + "TM", + "TN", + "TO", + "TP", + "TR", + "TT", + "TV", + "TW", + "TZ", + "UA", + "UG", + "US", + "UY", + "UZ", + "VA", + "VC", + "VE", + "VG", + "VI", + "VN", + "VU", + "WF", + "WS", + "YE", + "YT", + "ZA", + "ZM", + "ZW", +) + +# List of supported top-level domains that you can register with Amazon Route53 +AWS_SUPPORTED_TLDS = ( + "ac", + "academy", + "accountants", + "actor", + "adult", + "agency", + "airforce", + "apartments", + "associates", + "auction", + "audio", + "band", + "bargains", + "bet", + "bike", + "bingo", + "biz", + "black", + "blue", + "boutique", + "builders", + "business", + "buzz", + "cab", + "cafe", + "camera", + "camp", + "capital", + "cards", + "care", + "careers", + "cash", + "casino", + "catering", + "cc", + "center", + "ceo", + "chat", + "cheap", + "church", + "city", + "claims", + "cleaning", + "click", + "clinic", + "clothing", + "cloud", + "club", + "coach", + "codes", + "coffee", + "college", + "com", + "community", + "company", + "computer", + "condos", + "construction", + "consulting", + "contractors", + "cool", + "coupons", + "credit", + "creditcard", + "cruises", + "dance", + "dating", + "deals", + "degree", + "delivery", + "democrat", + "dental", + "diamonds", + "diet", + "digital", + "direct", + "directory", + "discount", + "dog", + "domains", + "education", + "email", + "energy", + "engineering", + "enterprises", + "equipment", + "estate", + "events", + "exchange", + "expert", + "exposed", + "express", + "fail", + "farm", + "finance", + "financial", + "fish", + "fitness", + "flights", + "florist", + "flowers", + "fm", + "football", + "forsale", + "foundation", + "fund", + "furniture", + "futbol", + "fyi", + "gallery", + "games", + "gift", + "gifts", + "gives", + "glass", + "global", + "gmbh", + "gold", + "golf", + "graphics", + "gratis", + "green", + "gripe", + "group", + "guide", + "guitars", + "guru", + "haus", + "healthcare", + "help", + "hiv", + "hockey", + "holdings", + "holiday", + "host", + "hosting", + "house", + "im", + "immo", + "immobilien", + "industries", + "info", + "ink", + "institute", + "insure", + "international", + "investments", + "io", + "irish", + "jewelry", + "juegos", + "kaufen", + "kim", + "kitchen", + "kiwi", + "land", + "lease", + "legal", + "lgbt", + "life", + "lighting", + "limited", + "limo", + "link", + "live", + "loan", + "loans", + "lol", + "maison", + "management", + "marketing", + "mba", + "media", + "memorial", + "mobi", + "moda", + "money", + "mortgage", + "movie", + "name", + "net", + "network", + "news", + "ninja", + "onl", + "online", + "org", + "partners", + "parts", + "photo", + "photography", + "photos", + "pics", + "pictures", + "pink", + "pizza", + "place", + "plumbing", + "plus", + "poker", + "porn", + "press", + "pro", + "productions", + "properties", + "property", + "pub", + "qpon", + "recipes", + "red", + "reise", + "reisen", + "rentals", + "repair", + "report", + "republican", + "restaurant", + "reviews", + "rip", + "rocks", + "run", + "sale", + "sarl", + "school", + "schule", + "services", + "sex", + "sexy", + "shiksha", + "shoes", + "show", + "singles", + "site", + "soccer", + "social", + "solar", + "solutions", + "space", + "store", + "studio", + "style", + "sucks", + "supplies", + "supply", + "support", + "surgery", + "systems", + "tattoo", + "tax", + "taxi", + "team", + "tech", + "technology", + "tennis", + "theater", + "tienda", + "tips", + "tires", + "today", + "tools", + "tours", + "town", + "toys", + "trade", + "training", + "tv", + "university", + "uno", + "vacations", + "vegas", + "ventures", + "vg", + "viajes", + "video", + "villas", + "vision", + "voyage", + "watch", + "website", + "wedding", + "wiki", + "wine", + "works", + "world", + "wtf", + "xyz", + "zone", +) + +VALID_DOMAIN_REGEX = re.compile( + r"(?:[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?\.)+[a-z0-9][a-z0-9-]{0,61}[a-z0-9]" +) +PHONE_NUMBER_REGEX = re.compile(r"\+\d*\.\d+$") + + +class DomainFilterField(str, Enum): + DOMAIN_NAME = "DomainName" + EXPIRY = "Expiry" + + +class DomainSortOrder(str, Enum): + ASCENDING = "ASC" + DESCENDING = "DES" + + +class DomainFilterOperator(str, Enum): + LE = "LE" + GE = "GE" + BEGINS_WITH = "BEGINS_WITH" + + +def is_valid_enum(value: Any, enum_cls: Type[Enum]) -> bool: + try: + enum_cls(value) + return True + except ValueError: + return False + + +class ValidationException(Exception): + def __init__(self, errors: List[str]): + super().__init__("\n\t".join(errors)) + self.errors = errors + + +class Route53DomainsOperation(BaseModel): + def __init__( + self, + id_: str, + domain_name: str, + status: str, + type_: str, + submitted_date: datetime, + last_updated_date: datetime, + message: Optional[str] = None, + status_flag: Optional[str] = None, + ): + self.id = id_ + self.domain_name = domain_name + self.status = status + self.type = type_ + self.submitted_date = submitted_date + self.last_updated_date = last_updated_date + self.message = message + self.status_flag = status_flag + + @classmethod + def validate( # type: ignore[misc,no-untyped-def] + cls, + domain_name: str, + status: str, + type_: str, + message: Optional[str] = None, + status_flag: Optional[str] = None, + ): + + id_ = str(mock_random.uuid4()) + submitted_date = datetime.now(timezone.utc) + last_updated_date = datetime.now(timezone.utc) + + return cls( + id_, + domain_name, + status, + type_, + submitted_date, + last_updated_date, + message, + status_flag, + ) + + def to_json(self) -> Dict[str, Any]: + d = { + "OperationId": self.id, + "Status": self.status, + "StatusFlag": self.status_flag, + "DomainName": self.domain_name, + "LastUpdatedDate": self.last_updated_date.timestamp(), + "SubmittedDate": self.submitted_date.timestamp(), + "Type": self.type, + } + + if self.message: + d["Message"] = self.message + + return d + + +class Route53DomainsContactDetail(BaseModel): + def __init__( + self, + address_line_1: Optional[str] = None, + address_line_2: Optional[str] = None, + city: Optional[str] = None, + contact_type: Optional[str] = None, + country_code: Optional[str] = None, + email: Optional[str] = None, + extra_params: Optional[List[Dict[str, Any]]] = None, + fax: Optional[str] = None, + first_name: Optional[str] = None, + last_name: Optional[str] = None, + organization_name: Optional[str] = None, + phone_number: Optional[str] = None, + state: Optional[str] = None, + zip_code: Optional[str] = None, + ): + super().__init__() + self.address_line_1 = address_line_1 + self.address_line_2 = address_line_2 + self.city = city + self.contact_type = contact_type + self.country_code = country_code + self.email = email + self.extra_params = extra_params + self.fax = fax + self.first_name = first_name + self.last_name = last_name + self.organization_name = organization_name + self.phone_number = phone_number + self.state = state + self.zip_code = zip_code + + @classmethod + def validate( # type: ignore[misc, no-untyped-def] + cls, + address_line_1: Optional[str] = None, + address_line_2: Optional[str] = None, + city: Optional[str] = None, + contact_type: Optional[str] = None, + country_code: Optional[str] = None, + email: Optional[str] = None, + extra_params: Optional[List[Dict[str, Any]]] = None, + fax: Optional[str] = None, + first_name: Optional[str] = None, + last_name: Optional[str] = None, + organization_name: Optional[str] = None, + phone_number: Optional[str] = None, + state: Optional[str] = None, + zip_code: Optional[str] = None, + ): + input_errors: List[str] = [] + + cls.__validate_str_len(address_line_1, "AddressLine1", 255, input_errors) + cls.__validate_str_len(address_line_2, "AddressLine2", 255, input_errors) + cls.__validate_str_len(city, "City", 255, input_errors) + cls.__validate_str_len(email, "Email", 255, input_errors) + cls.__validate_str_len(fax, "Fax", 255, input_errors) + cls.__validate_str_len(first_name, "FirstName", 255, input_errors) + cls.__validate_str_len(last_name, "LastName", 255, input_errors) + cls.__validate_str_len(state, "State", 255, input_errors) + cls.__validate_str_len(zip_code, "ZipCode", 255, input_errors) + + if contact_type: + if contact_type not in DOMAIN_CONTACT_DETAIL_CONTACT_TYPES: + input_errors.append(f"Invalid contact type {contact_type}") + else: + if contact_type != "PERSON" and not organization_name: + input_errors.append( + "Must supply OrganizationName when ContactType is not PERSON" + ) + + if country_code and country_code not in DOMAIN_CONTACT_DETAIL_COUNTRY_CODES: + input_errors.append(f"CountryCode {country_code} is invalid") + + if phone_number and not PHONE_NUMBER_REGEX.match(phone_number): + input_errors.append("PhoneNumber is in an invalid format") + + if input_errors: + raise ValidationException(input_errors) + + return cls( + address_line_1, + address_line_2, + city, + contact_type, + country_code, + email, + extra_params, + fax, + first_name, + last_name, + organization_name, + phone_number, + state, + zip_code, + ) + + @classmethod + def validate_dict(cls, d: Dict[str, Any]): # type: ignore[misc, no-untyped-def] + address_line_1 = d.get("AddressLine1") + address_line_2 = d.get("AddressLine2") + city = d.get("City") + contact_type = d.get("ContactType") + country_code = d.get("CountryCode") + email = d.get("Email") + extra_params = d.get("ExtraParams") + fax = d.get("Fax") + first_name = d.get("FirstName") + last_name = d.get("LastName") + organization_name = d.get("OrganizationName") + phone_number = d.get("PhoneNumber") + state = d.get("State") + zip_code = d.get("ZipCode") + return cls.validate( + address_line_1=address_line_1, + address_line_2=address_line_2, + city=city, + contact_type=contact_type, + country_code=country_code, + email=email, + extra_params=extra_params, + fax=fax, + first_name=first_name, + last_name=last_name, + organization_name=organization_name, + phone_number=phone_number, + state=state, + zip_code=zip_code, + ) + + @staticmethod + def __validate_str_len( + value: Optional[str], field_name: str, max_len: int, input_errors: List[str] + ) -> None: + if value and len(value) > max_len: + input_errors.append(f"Length of {field_name} is more than {max_len}") + + def to_json(self) -> Dict[str, Any]: + d = { + "FirstName": self.first_name, + "LastName": self.last_name, + "ContactType": self.contact_type, + "OrganizationName": self.organization_name, + "AddressLine1": self.address_line_1, + "AddressLine2": self.address_line_2, + "City": self.city, + "State": self.state, + "CountryCode": self.country_code, + "ZipCode": self.zip_code, + "PhoneNumber": self.phone_number, + "Email": self.email, + "Fax": self.fax, + "ExtraParams": self.extra_params, + } + + return {key: value for key, value in d.items() if value is not None} + + +class NameServer: + def __init__(self, name: str, glue_ips: List[str]): + self.name = name + self.glue_ips = glue_ips + + @classmethod + def validate(cls, name: str, glue_ips: Optional[List[str]] = None): # type: ignore[misc,no-untyped-def] + glue_ips = glue_ips or [] + input_errors: List[str] = [] + + if not VALID_DOMAIN_REGEX.match(name): + input_errors.append(f"{name} is not a valid host name") + + num_ipv4_addresses = 0 + num_ipv6_addresses = 0 + for ip in glue_ips: + try: + address = ip_address(ip) + if isinstance(address, IPv4Address): + num_ipv4_addresses += 1 + elif isinstance(address, IPv6Address): + num_ipv6_addresses += 1 + except ValueError: + input_errors.append(f"{ip} is not a valid IP address") + + if num_ipv4_addresses > 1: + input_errors.append("GlueIps list must include only 1 IPv4 address") + if num_ipv6_addresses > 1: + input_errors.append("GlueIps list must include only 1 IPv6 address") + + if input_errors: + raise ValidationException(input_errors) + + return cls(name, glue_ips) + + @classmethod + def validate_dict(cls, data: Dict[str, Any]): # type: ignore[misc,no-untyped-def] + name = data.get("Name") + glue_ips = data.get("GlueIps") + return cls.validate(name, glue_ips) # type: ignore[arg-type] + + def to_json(self) -> Dict[str, Any]: + d: Dict[str, Any] = {"Name": self.name} + if self.glue_ips: + d["GlueIps"] = self.glue_ips + return d + + +class Route53Domain(BaseModel): + def __init__( + self, + domain_name: str, + nameservers: List[NameServer], + auto_renew: bool, + admin_contact: Route53DomainsContactDetail, + registrant_contact: Route53DomainsContactDetail, + tech_contact: Route53DomainsContactDetail, + admin_privacy: bool, + registrant_privacy: bool, + tech_privacy: bool, + registrar_name: str, + whois_server: str, + registrar_url: str, + abuse_contact_email: str, + abuse_contact_phone: str, + registry_domain_id: str, + creation_date: datetime, + updated_date: datetime, + expiration_date: datetime, + reseller: str, + status_list: List[str], + dns_sec_keys: List[Dict[str, Any]], + extra_params: List[Dict[str, Any]], + ): + self.domain_name = domain_name + self.nameservers = nameservers + self.auto_renew = auto_renew + self.admin_contact = admin_contact + self.registrant_contact = registrant_contact + self.tech_contact = tech_contact + self.admin_privacy = admin_privacy + self.registrant_privacy = registrant_privacy + self.tech_privacy = tech_privacy + self.registrar_name = registrar_name + self.whois_server = whois_server + self.registrar_url = registrar_url + self.abuse_contact_email = abuse_contact_email + self.abuse_contact_phone = abuse_contact_phone + self.registry_domain_id = registry_domain_id + self.creation_date = creation_date + self.updated_date = updated_date + self.expiration_date = expiration_date + self.reseller = reseller + self.status_list = status_list + self.dns_sec_keys = dns_sec_keys + self.extra_params = extra_params + + @classmethod + def validate( # type: ignore[misc,no-untyped-def] + cls, + domain_name: str, + admin_contact: Route53DomainsContactDetail, + registrant_contact: Route53DomainsContactDetail, + tech_contact: Route53DomainsContactDetail, + nameservers: Optional[List[Dict[str, Any]]] = None, # type: ignore[type-arg] + auto_renew: bool = True, + admin_privacy: bool = True, + registrant_privacy: bool = True, + tech_privacy: bool = True, + registrar_name: Optional[str] = None, + whois_server: Optional[str] = None, + registrar_url: Optional[str] = None, + abuse_contact_email: Optional[str] = None, + abuse_contact_phone: Optional[str] = None, + registry_domain_id: Optional[str] = None, + expiration_date: Optional[datetime] = None, + reseller: Optional[str] = None, + dns_sec_keys: Optional[List[Dict[str, Any]]] = None, + extra_params: Optional[List[Dict[str, Any]]] = None, + ): + input_errors: List[str] = [] + + cls.validate_domain_name(domain_name, input_errors) + + nameservers = nameservers or [] + try: + nameservers = [ + NameServer.validate_dict(nameserver) for nameserver in nameservers + ] or [ + NameServer.validate(name="ns-2048.awscdn-64.net"), + NameServer.validate(name="ns-2051.awscdn-67.net"), + NameServer.validate(name="ns-2050.awscdn-66.net"), + NameServer.validate(name="ns-2049.awscdn-65.net"), + ] + except ValidationException as e: + input_errors += e.errors + + creation_date = datetime.now(timezone.utc) + updated_date = datetime.now(timezone.utc) + expiration_date = expiration_date or datetime.now(timezone.utc) + timedelta( + days=365 * 10 + ) + registrar_name = registrar_name or "GANDI SAS" + whois_server = whois_server or "whois.gandi.net" + registrar_url = registrar_url or "http://www.gandi.net" + abuse_contact_email = abuse_contact_email or "abuse@support.gandi.net" + status_list = ["SUCCEEDED"] + + time_until_expiration = expiration_date - datetime.now(timezone.utc) + if time_until_expiration < timedelta( + days=365 + ) or time_until_expiration > timedelta(days=365 * 10): + input_errors.append( + "ExpirationDate must by between 1 and 10 years from now" + ) + + if input_errors: + raise ValidationException(input_errors) + + return cls( + domain_name=domain_name, + nameservers=nameservers, # type: ignore[arg-type] + auto_renew=auto_renew, # type: ignore[arg-type] + admin_contact=admin_contact, + registrant_contact=registrant_contact, + tech_contact=tech_contact, + admin_privacy=admin_privacy, + registrant_privacy=registrant_privacy, + tech_privacy=tech_privacy, + registrar_name=registrar_name, + whois_server=whois_server, + registrar_url=registrar_url, + abuse_contact_email=abuse_contact_email, # type: ignore[arg-type] + abuse_contact_phone=abuse_contact_phone, # type: ignore[arg-type] + registry_domain_id=registry_domain_id, # type: ignore[arg-type] + creation_date=creation_date, # type: ignore[arg-type] + updated_date=updated_date, + expiration_date=expiration_date, + reseller=reseller, # type: ignore[arg-type] + status_list=status_list, # type: ignore[arg-type] + dns_sec_keys=dns_sec_keys, # type: ignore[arg-type] + extra_params=extra_params, # type: ignore[arg-type] + ) + + @staticmethod + def validate_domain_name(domain_name: str, input_errors: List[str]) -> None: + if not VALID_DOMAIN_REGEX.match(domain_name): + input_errors.append("Invalid domain name") + return + + tld = domain_name.split(".")[-1] + if tld not in AWS_SUPPORTED_TLDS: + raise UnsupportedTLDException(tld) + + def to_json(self) -> Dict[str, Any]: + return { + "DomainName": self.domain_name, + "Nameservers": [nameserver.to_json() for nameserver in self.nameservers], + "AutoRenew": self.auto_renew, + "AdminContact": self.admin_contact.to_json(), + "RegistrantContact": self.registrant_contact.to_json(), + "TechContact": self.tech_contact.to_json(), + "AdminPrivacy": self.admin_privacy, + "RegistrantPrivacy": self.registrant_privacy, + "TechPrivacy": self.tech_privacy, + "RegistrarName": self.registrar_name, + "WhoIsServer": self.whois_server, + "RegistrarUrl": self.registrar_url, + "AbuseContactEmail": self.abuse_contact_email, + "AbuseContactPhone": self.abuse_contact_phone, + "RegistryDomainId": "", + "CreationDate": self.creation_date.timestamp(), + "UpdateDate": self.updated_date.timestamp(), + "ExpirationDate": self.expiration_date.timestamp(), + "Reseller": self.reseller, + "DnsSec": "", + "StatusList": self.status_list, + "DnsSecKeys": self.dns_sec_keys, + "BillingContact": self.admin_contact.to_json(), + } + + +class DomainsFilter: + def __init__( + self, name: DomainFilterField, operator: DomainFilterOperator, values: List[str] + ): + self.name: DomainFilterField = name + self.operator: DomainFilterOperator = operator + self.values = values + + def filter(self, domain: Route53Domain) -> bool: + if self.name == DomainFilterField.DOMAIN_NAME: + return self.__filter_by_domain_name(domain) + return self.__filter_by_expiry_date(domain) + + def __filter_by_domain_name(self, domain: Route53Domain) -> bool: + return any([domain.domain_name.startswith(value) for value in self.values]) + + def __filter_by_expiry_date(self, domain: Route53Domain) -> bool: + return ( + any( + [ + value + for value in self.values + if domain.expiration_date + >= datetime.fromtimestamp(float(value), tz=timezone.utc) + ] + ) + if self.operator == DomainFilterOperator.GE + else any( + [ + value + for value in self.values + if domain.expiration_date + <= datetime.fromtimestamp(float(value), tz=timezone.utc) + ] + ) + ) + + @classmethod + def validate(cls, name: str, operator: str, values: List[str]): # type: ignore[misc, no-untyped-def] + input_errors: List[str] = [] + + if not is_valid_enum(name, DomainFilterField): + input_errors.append(f"Cannot filter by field {name}") + + if not is_valid_enum(operator, DomainFilterOperator): + input_errors.append(f"Invalid filter operator {operator}") + + if len(values) != 1: + input_errors.append("Multiple filter values are not currently supported") + + if ( + name == DomainFilterField.DOMAIN_NAME + and operator != DomainFilterOperator.BEGINS_WITH + ): + input_errors.append( + f"Operator {operator} cannot be used with the DomainName filter" + ) + + if name == DomainFilterField.EXPIRY and operator not in ( + DomainFilterOperator.GE, + DomainFilterOperator.LE, + ): + input_errors.append( + f"Operator {operator} cannot be used with the Expiry filter" + ) + + if input_errors: + raise ValidationException(input_errors) + + return cls( + name=DomainFilterField(name), + operator=DomainFilterOperator(operator), + values=values, + ) + + @classmethod + def validate_dict(cls, data: Dict[str, Any]): # type: ignore[misc,no-untyped-def] + name = data.get("Name") + operator = data.get("Operator") + values = data.get("Values") + return cls.validate(name=name, operator=operator, values=values) # type: ignore[arg-type] + + +class DomainsSortCondition: + def __init__(self, name: DomainFilterField, sort_order: DomainSortOrder): + self.name: DomainFilterField = name + self.sort_order: DomainSortOrder = sort_order + + @classmethod + def validate(cls, name: str, sort_order: str): # type: ignore[misc,no-untyped-def] + input_errors: List[str] = [] + if not is_valid_enum(name, DomainFilterField): + input_errors.append(f"Cannot sort by field {name}") + + if not is_valid_enum(sort_order, DomainSortOrder): + input_errors.append(f"Invalid sort order {sort_order}") + + if input_errors: + raise ValidationException(input_errors) + + return cls(name=DomainFilterField(name), sort_order=DomainSortOrder(sort_order)) + + @classmethod + def validate_dict(cls, data: Dict[str, Any]): # type: ignore[misc,no-untyped-def] + name = data.get("Name") + sort_order = data.get("SortOrder") + return cls.validate(name=name, sort_order=sort_order) # type: ignore[arg-type] diff --git a/tests/test_route53domains/__init__.py b/tests/test_route53domains/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/tests/test_route53domains/test_route53domains_domain.py b/tests/test_route53domains/test_route53domains_domain.py new file mode 100644 index 000000000..891329b32 --- /dev/null +++ b/tests/test_route53domains/test_route53domains_domain.py @@ -0,0 +1,562 @@ +from datetime import datetime, timedelta, timezone +from typing import Dict, List + +import boto3 +import pytest +from botocore.exceptions import ClientError + +from moto import mock_aws + + +@pytest.fixture(name="domain_parameters") +def generate_domain_parameters() -> Dict: + return { + "DomainName": "domain.com", + "DurationInYears": 3, + "AutoRenew": True, + "AdminContact": { + "FirstName": "First", + "LastName": "Last", + "ContactType": "PERSON", + "AddressLine1": "address 1", + "AddressLine2": "address 2", + "City": "New York City", + "CountryCode": "US", + "ZipCode": "123123123", + "Email": "email@gmail.com", + "Fax": "+1.1234567890", + }, + "RegistrantContact": { + "FirstName": "First", + "LastName": "Last", + "ContactType": "PERSON", + "AddressLine1": "address 1", + "AddressLine2": "address 2", + "City": "New York City", + "CountryCode": "US", + "ZipCode": "123123123", + "Email": "email@gmail.com", + "Fax": "+1.1234567890", + }, + "TechContact": { + "FirstName": "First", + "LastName": "Last", + "ContactType": "PERSON", + "AddressLine1": "address 1", + "AddressLine2": "address 2", + "City": "New York City", + "CountryCode": "US", + "ZipCode": "123123123", + "Email": "email@gmail.com", + "Fax": "+1.1234567890", + }, + "PrivacyProtectAdminContact": True, + "PrivacyProtectRegistrantContact": True, + "PrivacyProtectTechContact": True, + } + + +@pytest.fixture(name="invalid_domain_parameters") +def generate_invalid_domain_parameters(domain_parameters: Dict) -> Dict: + domain_parameters["DomainName"] = "a" + domain_parameters["DurationInYears"] = 500 + return domain_parameters + + +@mock_aws +def test_register_domain(domain_parameters: Dict): + route53domains_client = boto3.client("route53domains", region_name="global") + res = route53domains_client.register_domain(**domain_parameters) + + operation_id = res["OperationId"] + + operations = route53domains_client.list_operations(Type=["REGISTER_DOMAIN"])[ + "Operations" + ] + for operation in operations: + if operation["OperationId"] == operation_id: + return + + assert operation_id in [ + operation["OperationId"] for operation in operations + ], "Could not find expected operation id returned from `register_domain` in operation list" + + +@mock_aws +def test_register_domain_creates_hosted_zone( + domain_parameters: Dict, +): + """Test good register domain API calls.""" + route53domains_client = boto3.client("route53domains", region_name="global") + route53_client = boto3.client("route53", region_name="global") + route53domains_client.register_domain(**domain_parameters) + + res = route53_client.list_hosted_zones() + assert "domain.com" in [ + zone["Name"] for zone in res["HostedZones"] + ], "`register_domain` did not create a new hosted zone with the same name" + + +@mock_aws +def test_register_domain_fails_on_invalid_input( + invalid_domain_parameters: Dict, +): + route53domains_client = boto3.client("route53domains", region_name="global") + route53_client = boto3.client("route53", region_name="global") + with pytest.raises(ClientError) as exc: + route53domains_client.register_domain(**invalid_domain_parameters) + + err = exc.value.response["Error"] + assert err["Code"] == "InvalidInput" + + res = route53_client.list_hosted_zones() + assert len(res["HostedZones"]) == 0 + + +@mock_aws +def test_register_domain_fails_on_invalid_tld(domain_parameters: Dict): + route53domains_client = boto3.client("route53domains", region_name="global") + route53_client = boto3.client("route53", region_name="global") + + params = domain_parameters.copy() + params["DomainName"] = "test.non-existing-tld" + with pytest.raises(ClientError) as exc: + route53domains_client.register_domain(**params) + + err = exc.value.response["Error"] + assert err["Code"] == "UnsupportedTLD" + + res = route53_client.list_hosted_zones() + assert len(res["HostedZones"]) == 0 + + +@mock_aws +def test_list_operations(domain_parameters: Dict): + route53domains_client = boto3.client("route53domains", region_name="global") + route53domains_client.register_domain(**domain_parameters) + operations = route53domains_client.list_operations()["Operations"] + assert len(operations) == 1 + + future_time = datetime.now(timezone.utc) + timedelta(minutes=1) + operations = route53domains_client.list_operations( + SubmittedSince=future_time.timestamp() + )["Operations"] + assert len(operations) == 0 + + operations = route53domains_client.list_operations(Status=["SUCCESSFUL"])[ + "Operations" + ] + assert len(operations) == 1 + + operations = route53domains_client.list_operations(Status=["IN_PROGRESS"])[ + "Operations" + ] + assert len(operations) == 0 + + operations = route53domains_client.list_operations(Type=["REGISTER_DOMAIN"])[ + "Operations" + ] + assert len(operations) == 1 + + operations = route53domains_client.list_operations(Type=["DELETE_DOMAIN"])[ + "Operations" + ] + assert len(operations) == 0 + + +@mock_aws +def test_list_operations_invalid_input(): + route53domains_client = boto3.client("route53domains", region_name="global") + with pytest.raises(ClientError) as exc: + _ = route53domains_client.list_operations(Type=["INVALID_TYPE"])["Operations"] + + err = exc.value.response["Error"] + assert err["Code"] == "InvalidInput" + + with pytest.raises(ClientError) as exc: + _ = route53domains_client.list_operations(Status=["INVALID_STATUS"])[ + "Operations" + ] + err = exc.value.response["Error"] + assert err["Code"] == "InvalidInput" + + +@mock_aws +def test_get_operation_detail(domain_parameters: Dict): + route53domains_client = boto3.client("route53domains", region_name="global") + res = route53domains_client.register_domain(**domain_parameters) + expected_operation_id = res["OperationId"] + operation = route53domains_client.get_operation_detail( + OperationId=expected_operation_id + ) + assert operation["OperationId"] == expected_operation_id + assert operation["Status"] == "SUCCESSFUL" + assert operation["Type"] == "REGISTER_DOMAIN" + + +@mock_aws +def test_get_nonexistent_operation_detail(): + route53domains_client = boto3.client("route53domains", region_name="global") + with pytest.raises(ClientError) as exc: + route53domains_client.get_operation_detail( + OperationId="non-exiting-operation-id" + ) + + err = exc.value.response["Error"] + assert err["Code"] == "InvalidInput" + + +@mock_aws +def test_duplicate_requests(domain_parameters: Dict): + route53domains_client = boto3.client("route53domains", region_name="global") + route53domains_client.register_domain(**domain_parameters) + with pytest.raises(ClientError) as exc: + route53domains_client.register_domain(**domain_parameters) + err = exc.value.response["Error"] + assert err["Code"] == "DuplicateRequest" + + +@mock_aws +def test_domain_limit(domain_parameters: Dict): + route53domains_client = boto3.client("route53domains", region_name="global") + params = domain_parameters.copy() + for i in range(20): + params["DomainName"] = f"domain-{i}.com" + route53domains_client.register_domain(**params) + + params["DomainName"] = "domain-20.com" + with pytest.raises(ClientError) as exc: + route53domains_client.register_domain(**params) + + err = exc.value.response["Error"] + assert err["Code"] == "DomainLimitExceeded" + + +@mock_aws +def test_get_domain_detail(domain_parameters: Dict): + route53domains_client = boto3.client("route53domains", region_name="global") + route53domains_client.register_domain(**domain_parameters) + res = route53domains_client.get_domain_detail( + DomainName=domain_parameters["DomainName"] + ) + assert res["DomainName"] == domain_parameters["DomainName"] + + +@mock_aws +def test_get_invalid_domain_detail(domain_parameters): + route53domains_client = boto3.client("route53domains", region_name="global") + route53domains_client.register_domain(**domain_parameters) + + with pytest.raises(ClientError) as exc: + route53domains_client.get_domain_detail(DomainName="not-a-domain") + + err = exc.value.response["Error"] + assert err["Code"] == "InvalidInput" + + with pytest.raises(ClientError) as exc: + route53domains_client.get_domain_detail(DomainName="test.non-existing-tld") + + err = exc.value.response["Error"] + assert err["Code"] == "UnsupportedTLD" + + +@mock_aws +def test_list_domains(domain_parameters: Dict): + route53domains_client = boto3.client("route53domains", region_name="global") + route53domains_client.register_domain(**domain_parameters) + res = route53domains_client.list_domains() + + assert len(res["Domains"]) == 1 + params = domain_parameters.copy() + params["DomainName"] = "new-domain.com" + route53domains_client.register_domain(**params) + res = route53domains_client.list_domains() + assert len(res["Domains"]) == 2 + + +@mock_aws +@pytest.mark.parametrize( + "filters,expected_domains_len", + [ + ( + [ + { + "Name": "DomainName", + "Operator": "BEGINS_WITH", + "Values": ["some-non-registered-domain.com"], + } + ], + 0, # expected_domains_len + ), + ( + [ + { + "Name": "DomainName", + "Operator": "BEGINS_WITH", + "Values": ["domain.com"], + } + ], + 1, # expected_domains_len + ), + ( + [ + { + "Name": "Expiry", + "Operator": "GE", + "Values": [ + str( + datetime.fromisocalendar( + year=2012, week=20, day=3 + ).timestamp() + ) + ], + } + ], + 1, # expected_domains_len + ), + ( + [ + { + "Name": "Expiry", + "Operator": "GE", + "Values": [ + str( + datetime.fromisocalendar( + year=2050, week=20, day=3 + ).timestamp() + ) + ], + } + ], + 0, # expected_domains_len + ), + ], +) +def test_list_domains_filters( + domain_parameters: Dict, filters: List[Dict], expected_domains_len: int +): + route53domains_client = boto3.client("route53domains", region_name="global") + route53domains_client.register_domain(**domain_parameters) + res = route53domains_client.list_domains(FilterConditions=filters) + assert len(res["Domains"]) == expected_domains_len + + +@mock_aws +def test_list_domains_sort_condition(domain_parameters: Dict): + route53domains_client = boto3.client("route53domains", region_name="global") + params = domain_parameters.copy() + params["DomainName"] = "adomain.com" + route53domains_client.register_domain(**params) + params["DomainName"] = "bdomain.com" + route53domains_client.register_domain(**params) + sort = {"Name": "DomainName", "SortOrder": "DES"} + res = route53domains_client.list_domains(SortCondition=sort) + domains = res["Domains"] + assert domains[0]["DomainName"] == "bdomain.com" + assert domains[1]["DomainName"] == "adomain.com" + + sort = {"Name": "Expiry", "SortOrder": "ASC"} + res = route53domains_client.list_domains(SortCondition=sort) + domains = res["Domains"] + assert domains[0]["DomainName"] == "adomain.com" + assert domains[1]["DomainName"] == "bdomain.com" + + +@mock_aws +def test_list_domains_invalid_filter(domain_parameters: Dict): + route53domains_client = boto3.client("route53domains", region_name="global") + route53domains_client.register_domain(**domain_parameters) + filters = [ + { + "Name": "InvalidField", + "Operator": "InvalidOperator", + "Values": ["value-1", "value-2"], # multiple values isn't supported + } + ] + + with pytest.raises(ClientError) as exc: + route53domains_client.list_domains(FilterConditions=filters) + + err = exc.value.response["Error"] + assert err["Code"] == "InvalidInput" + + +@mock_aws +def test_list_domains_invalid_sort_condition(domain_parameters: Dict): + route53domains_client = boto3.client("route53domains", region_name="global") + route53domains_client.register_domain(**domain_parameters) + sort = { + "Name": "InvalidField", + "SortOrder": "InvalidOrder", + } + + with pytest.raises(ClientError) as exc: + route53domains_client.list_domains(SortCondition=sort) + + err = exc.value.response["Error"] + assert err["Code"] == "InvalidInput" + + +@mock_aws +def test_list_domains_sort_condition_not_the_same_as_filter_condition( + domain_parameters: Dict, +): + route53domains_client = boto3.client("route53domains", region_name="global") + route53domains_client.register_domain(**domain_parameters) + sort = { + "Name": "Expiry", + "SortOrder": "ASC", + } + filters = [ + { + "Name": "DomainName", + "Operator": "BEGINS_WITH", + "Values": ["domain.com"], + } + ] + + with pytest.raises(ClientError) as exc: + route53domains_client.list_domains(FilterConditions=filters, SortCondition=sort) + + err = exc.value.response["Error"] + assert err["Code"] == "InvalidInput" + + +@mock_aws +def test_delete_domain(domain_parameters: Dict): + route53domains_client = boto3.client("route53domains", region_name="global") + route53domains_client.register_domain(**domain_parameters) + domains = route53domains_client.list_domains()["Domains"] + assert len(domains) == 1 + route53domains_client.delete_domain(DomainName=domain_parameters["DomainName"]) + domains = route53domains_client.list_domains()["Domains"] + assert len(domains) == 0 + operations = route53domains_client.list_operations(Type=["DELETE_DOMAIN"])[ + "Operations" + ] + assert len(operations) == 1 + + +@mock_aws +def test_delete_invalid_domain(domain_parameters: Dict): + route53domains_client = boto3.client("route53domains", region_name="global") + domains = route53domains_client.list_domains()["Domains"] + assert len(domains) == 0 + with pytest.raises(ClientError) as exc: + route53domains_client.delete_domain(DomainName=domain_parameters["DomainName"]) + + err = exc.value.response["Error"] + assert err["Code"] == "InvalidInput" + + +@mock_aws +@pytest.mark.parametrize( + "nameservers", + [ + [{"Name": "1-nameserver.net"}, {"Name": "2-nameserver.net"}], + [ + {"Name": "3-nameserver.net", "GlueIps": ["1.1.1.2"]}, + {"Name": "4-nameserver.net", "GlueIps": ["1.1.1.1"]}, + ], + ], +) +def test_update_domain_nameservers(domain_parameters: Dict, nameservers: List[Dict]): + route53domains_client = boto3.client("route53domains", region_name="global") + route53domains_client.register_domain(**domain_parameters) + operation_id = route53domains_client.update_domain_nameservers( + DomainName=domain_parameters["DomainName"], Nameservers=nameservers + )["OperationId"] + domain = route53domains_client.get_domain_detail( + DomainName=domain_parameters["DomainName"] + ) + assert domain["Nameservers"] == nameservers + operation = route53domains_client.get_operation_detail(OperationId=operation_id) + assert operation["Type"] == "UPDATE_NAMESERVER" + assert operation["Status"] == "SUCCESSFUL" + + +@mock_aws +@pytest.mark.parametrize( + "nameservers", + [ + [{"Name": "1-nameserver.net", "GlueIps": ["1.1.1.1", "1.1.1.2"]}], + [ + { + "Name": "1-nameserver.net", + "GlueIps": [ + "2001:0db8:85a3:0000:0000:8a2e:0370:7334", + "2001:0db8:85a3:0000:0000:8a2e:0370:7334", + ], + } + ], + [ + { + "Name": "1-nameserver.net", + "GlueIps": [ + "2001:0db8:85a3:0000:0000:8a2e:0370:7334", + "2001:0db8:85a3:0000:0000:8a2e:0370:7334", + "1.1.1.1", + ], + } + ], + [ + { + "Name": "1-nameserver.net", + "GlueIps": [ + "2001:0db8:85a3:0000:0000:8a2e:0370:7334", + "2001:0db8:85a3:0000:0000:8a2e:0370:7334", + "1.1.1.1", + "1.1.1.2", + ], + } + ], + [ + { + "Name": "1-nameserver.net", + "GlueIps": [ + "2001:0db8:85a3:0000:0000:8a2e:0370:7334", + "1.1.1.1", + "not-an-ip-address", + ], + } + ], + ], +) +def test_update_domain_nameservers_with_multiple_glue_ips( + domain_parameters: Dict, nameservers: List[Dict] +): + route53domains_client = boto3.client("route53domains", region_name="global") + route53domains_client.register_domain(**domain_parameters) + with pytest.raises(ClientError) as exc: + route53domains_client.update_domain_nameservers( + DomainName=domain_parameters["DomainName"], Nameservers=nameservers + ) + err = exc.value.response["Error"] + assert err["Code"] == "InvalidInput" + + +@mock_aws +def test_update_domain_nameservers_requires_glue_ips(domain_parameters: Dict): + route53domains_client = boto3.client("route53domains", region_name="global") + route53domains_client.register_domain(**domain_parameters) + domain_name = domain_parameters["DomainName"] + nameservers = [{"Name": f"subdomain.{domain_name}"}] + with pytest.raises(ClientError) as exc: + route53domains_client.update_domain_nameservers( + DomainName=domain_parameters["DomainName"], Nameservers=nameservers + ) + err = exc.value.response["Error"] + assert err["Code"] == "InvalidInput" + + +@mock_aws +def test_update_domain_nameservers_for_nonexistent_domain(): + route53domains_client = boto3.client("route53domains", region_name="global") + nameservers = [{"Name": "1-nameserver.net"}, {"Name": "2-nameserver.net"}] + + with pytest.raises(ClientError) as exc: + route53domains_client.update_domain_nameservers( + DomainName="non-existent-domain.com", Nameservers=nameservers + ) + err = exc.value.response["Error"] + assert err["Code"] == "InvalidInput"