diff --git a/moto/ec2/exceptions.py b/moto/ec2/exceptions.py index c4652763eb0..bc934cd7dd7 100644 --- a/moto/ec2/exceptions.py +++ b/moto/ec2/exceptions.py @@ -239,6 +239,14 @@ def __init__(self, route_table_id, cidr): ) +class RouteAlreadyExistsError(EC2ClientError): + def __init__(self, cidr): + super().__init__( + "RouteAlreadyExists", + "The route identified by {0} already exists".format(cidr), + ) + + class InvalidInstanceIdError(EC2ClientError): def __init__(self, instance_id): if isinstance(instance_id, str): diff --git a/moto/ec2/models/route_tables.py b/moto/ec2/models/route_tables.py index 376eed1edee..94738372473 100644 --- a/moto/ec2/models/route_tables.py +++ b/moto/ec2/models/route_tables.py @@ -1,4 +1,5 @@ import ipaddress + from moto.core import get_account_id, CloudFormationModel from .core import TaggedEC2Resource from ..exceptions import ( @@ -7,6 +8,7 @@ InvalidRouteTableIdError, InvalidAssociationIdError, InvalidDestinationCIDRBlockParameterError, + RouteAlreadyExistsError, ) from ..utils import ( EC2_RESOURCE_TO_PREFIX, @@ -323,11 +325,10 @@ def create_route( elif EC2_RESOURCE_TO_PREFIX["vpc-endpoint"] in gateway_id: gateway = self.get_vpc_end_point(gateway_id) - try: - if destination_cidr_block: - ipaddress.IPv4Network(str(destination_cidr_block), strict=False) - except ValueError: - raise InvalidDestinationCIDRBlockParameterError(destination_cidr_block) + if destination_cidr_block: + self.__validate_destination_cidr_block( + destination_cidr_block, route_table + ) if nat_gateway_id is not None: nat_gateway = self.nat_gateways.get(nat_gateway_id) @@ -440,3 +441,25 @@ def delete_route( if not deleted: raise InvalidRouteError(route_table_id, cidr) return deleted + + def __validate_destination_cidr_block(self, destination_cidr_block, route_table): + """ + Utility function to check the destination CIDR block + Will validate the format and check for overlap with existing routes + """ + try: + ip_v4_network = ipaddress.IPv4Network( + str(destination_cidr_block), strict=False + ) + except ValueError: + raise InvalidDestinationCIDRBlockParameterError(destination_cidr_block) + + if not route_table.routes: + return + for route in route_table.routes.values(): + if not route.destination_cidr_block: + continue + if not route.local and ip_v4_network.overlaps( + ipaddress.IPv4Network(str(route.destination_cidr_block)) + ): + raise RouteAlreadyExistsError(destination_cidr_block) diff --git a/tests/test_ec2/test_route_tables.py b/tests/test_ec2/test_route_tables.py index c772fb2b8db..2f579b2c2bb 100644 --- a/tests/test_ec2/test_route_tables.py +++ b/tests/test_ec2/test_route_tables.py @@ -520,6 +520,56 @@ def get_target_route(): ex.value.response["Error"]["Code"].should.equal("InvalidRouteTableID.NotFound") +@mock_ec2 +def test_routes_already_exist(): + client = boto3.client("ec2", region_name="us-east-1") + ec2 = boto3.resource("ec2", region_name="us-east-1") + vpc = ec2.create_vpc(CidrBlock="10.0.0.0/16") + + main_route_table_id = client.describe_route_tables( + Filters=[ + {"Name": "vpc-id", "Values": [vpc.id]}, + {"Name": "association.main", "Values": ["true"]}, + ] + )["RouteTables"][0]["RouteTableId"] + main_route_table = ec2.RouteTable(main_route_table_id) + ROUTE_CIDR = "10.0.0.0/23" + ROUTE_SUB_CIDR = "10.0.0.0/24" + ROUTE_NO_CONFLICT_CIDR = "10.0.2.0/24" + + # Various route targets + igw = ec2.create_internet_gateway() + + # Create initial route + main_route_table.create_route(DestinationCidrBlock=ROUTE_CIDR, GatewayId=igw.id) + main_route_table.create_route( + DestinationCidrBlock=ROUTE_NO_CONFLICT_CIDR, GatewayId=igw.id + ) + + # Create + with pytest.raises(ClientError) as ex: + client.create_route( + RouteTableId=main_route_table.id, + DestinationCidrBlock=ROUTE_CIDR, + GatewayId=igw.id, + ) + + ex.value.response["ResponseMetadata"]["HTTPStatusCode"].should.equal(400) + ex.value.response["ResponseMetadata"].should.have.key("RequestId") + ex.value.response["Error"]["Code"].should.equal("RouteAlreadyExists") + + with pytest.raises(ClientError) as ex: + client.create_route( + RouteTableId=main_route_table.id, + DestinationCidrBlock=ROUTE_SUB_CIDR, + GatewayId=igw.id, + ) + + ex.value.response["ResponseMetadata"]["HTTPStatusCode"].should.equal(400) + ex.value.response["ResponseMetadata"].should.have.key("RequestId") + ex.value.response["Error"]["Code"].should.equal("RouteAlreadyExists") + + @mock_ec2 def test_routes_not_supported(): client = boto3.client("ec2", region_name="us-east-1")