feature: handle overlapping routes (#5187)

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):

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)

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")