EC2: vpc endpoint as a route target (#6316)

moto/ec2/exceptions.py

@@ -234,6 +234,15 @@ class RouteAlreadyExistsError(EC2ClientError):
         )
 
 
+class RouteNotSupportedError(EC2ClientError):
+    def __init__(self, vpce_id: str):
+        super().__init__(
+            "RouteNotSupported",
+            f"Route table contains unsupported route target: {vpce_id}. "
+            f"VPC Endpoints of this type cannot be used as route targets.",
+        )
+
+
 class InvalidInstanceIdError(EC2ClientError):
     def __init__(self, instance_id: Any):
         if isinstance(instance_id, str):

moto/ec2/models/route_tables.py

@@ -19,6 +19,7 @@ from ..exceptions import (
     InvalidAssociationIdError,
     InvalidDestinationCIDRBlockParameterError,
     RouteAlreadyExistsError,
+    RouteNotSupportedError,
 )
 from ..utils import (
     EC2_RESOURCE_TO_PREFIX,
@@ -141,6 +142,7 @@ class Route(CloudFormationModel):
         interface: Optional[NetworkInterface] = None,
         vpc_pcx: Optional[VPCPeeringConnection] = None,
         carrier_gateway: Optional[CarrierGateway] = None,
+        vpc_endpoint_id: Optional[str] = None,
     ):
         self.id = generate_route_id(
             route_table.id,
@@ -161,6 +163,7 @@ class Route(CloudFormationModel):
         self.interface = interface
         self.vpc_pcx = vpc_pcx
         self.carrier_gateway = carrier_gateway
+        self.vpc_endpoint_id = vpc_endpoint_id
 
     @property
     def physical_resource_id(self) -> str:
@@ -359,6 +362,7 @@ class RouteBackend:
         interface_id: Optional[str] = None,
         vpc_peering_connection_id: Optional[str] = None,
         carrier_gateway_id: Optional[str] = None,
+        vpc_endpoint_id: Optional[str] = None,
     ) -> Route:
         gateway = None
         nat_gateway = None
@@ -368,6 +372,13 @@ class RouteBackend:
         destination_prefix_list = None
         carrier_gateway = None
 
+        if vpc_endpoint_id:
+            vpce = self.describe_vpc_endpoints(vpc_end_point_ids=[vpc_endpoint_id])  # type: ignore[attr-defined]
+            if not vpce[0].endpoint_type == "GatewayLoadBalancer":
+                # https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/ec2/client/create_route.html
+                # VpcEndpointId (string) – The ID of a VPC endpoint. Supported for Gateway Load Balancer endpoints only.
+                raise RouteNotSupportedError(vpc_endpoint_id)
+
         route_table = self.get_route_table(route_table_id)
 
         if interface_id:
@@ -414,6 +425,7 @@ class RouteBackend:
             transit_gateway=transit_gateway,
             interface=interface,
             carrier_gateway=carrier_gateway,
+            vpc_endpoint_id=vpc_endpoint_id,
             vpc_pcx=self.get_vpc_peering_connection(vpc_peering_connection_id)  # type: ignore[attr-defined]
             if vpc_peering_connection_id
             else None,

moto/ec2/responses/route_tables.py

@@ -25,6 +25,7 @@ class RouteTables(EC2BaseResponse):
         interface_id = self._get_param("NetworkInterfaceId")
         pcx_id = self._get_param("VpcPeeringConnectionId")
         carrier_gateway_id = self._get_param("CarrierGatewayId")
+        vpc_endpoint_id = self._get_param("VpcEndpointId")
 
         self.ec2_backend.create_route(
             route_table_id,
@@ -39,6 +40,7 @@ class RouteTables(EC2BaseResponse):
             interface_id=interface_id,
             vpc_peering_connection_id=pcx_id,
             carrier_gateway_id=carrier_gateway_id,
+            vpc_endpoint_id=vpc_endpoint_id,
         )
 
         template = self.response_template(CREATE_ROUTE_RESPONSE)
@@ -212,6 +214,11 @@ DESCRIBE_ROUTE_TABLES_RESPONSE = """
                   <origin>CreateRoute</origin>
                   <state>active</state>
                 {% endif %}
+                {% if route.vpc_endpoint_id %}
+                  <gatewayId>{{ route.vpc_endpoint_id }}</gatewayId>
+                  <origin>CreateRoute</origin>
+                  <state>active</state>
+                {% endif %}
                 {% if route.instance %}
                   <instanceId>{{ route.instance.id }}</instanceId>
                   <origin>CreateRoute</origin>

tests/test_ec2/test_route_tables.py

@@ -1041,6 +1041,62 @@ def test_create_route_with_unknown_egress_only_igw():
     err["Message"].should.equal("The eigw ID 'eoigw' does not exist")
 
 
+@mock_ec2
+def test_create_route_with_vpc_endpoint():
+    # Setup
+    _, ec2_client, route_table, vpc = setup_vpc()
+    dest_cidr = "0.0.0.0/0"
+    vpc_end_point = ec2_client.create_vpc_endpoint(
+        VpcId=vpc.id,
+        ServiceName="com.amazonaws.vpce.eu-central-1.vpce-svc-084fa044c50cb1290",
+        RouteTableIds=[route_table.id],
+        VpcEndpointType="GatewayLoadBalancer",
+    )
+    vpce_id = vpc_end_point["VpcEndpoint"]["VpcEndpointId"]
+
+    # Execute
+    ec2_client.create_route(
+        DestinationCidrBlock=dest_cidr,
+        VpcEndpointId=vpce_id,
+        RouteTableId=route_table.id,
+    )
+    rt = ec2_client.describe_route_tables()
+    new_route = rt["RouteTables"][-1]["Routes"][1]
+
+    # Verify
+    assert new_route["DestinationCidrBlock"] == dest_cidr
+    assert new_route["GatewayId"] == vpce_id
+
+
+@mock_ec2
+def test_create_route_with_invalid_vpc_endpoint():
+    # Setup
+    _, ec2_client, route_table, vpc = setup_vpc()
+    dest_cidr = "0.0.0.0/0"
+    vpc_end_point = ec2_client.create_vpc_endpoint(
+        VpcId=vpc.id,
+        ServiceName="com.amazonaws.us-east-1.s3",
+        RouteTableIds=[route_table.id],
+        VpcEndpointType="Gateway",
+    )
+    vpce_id = vpc_end_point["VpcEndpoint"]["VpcEndpointId"]
+
+    # Execute
+    with pytest.raises(ClientError) as ex:
+        ec2_client.create_route(
+            DestinationCidrBlock=dest_cidr,
+            VpcEndpointId=vpce_id,
+            RouteTableId=route_table.id,
+        )
+    # Verify
+    err = ex.value.response["Error"]
+    assert err["Code"] == "RouteNotSupported"
+    assert (
+        err["Message"] == f"Route table contains unsupported route target: {vpce_id}. "
+        "VPC Endpoints of this type cannot be used as route targets."
+    )
+
+
 @mock_ec2
 def test_associate_route_table_by_gateway():
     ec2 = boto3.client("ec2", region_name="us-west-1")
@@ -1087,3 +1143,17 @@ def test_associate_route_table_by_subnet():
     verify[0]["Associations"][0]["SubnetId"].should.equals(subnet_id)
     verify[0]["Associations"][0]["RouteTableAssociationId"].should.equal(assoc_id)
     verify[0]["Associations"][0].doesnt.have.key("GatewayId")
+
+
+def setup_vpc():
+    ec2_resource = boto3.resource("ec2", region_name="eu-central-1")
+    ec2_client = boto3.client("ec2", region_name="eu-central-1")
+
+    vpc = ec2_resource.create_vpc(CidrBlock="10.0.0.0/16")
+    ec2_resource.create_subnet(
+        VpcId=vpc.id, CidrBlock="10.0.0.0/24", AvailabilityZone="us-west-2a"
+    )
+
+    route_table = ec2_resource.create_route_table(VpcId=vpc.id)
+
+    return ec2_resource, ec2_client, route_table, vpc