diff --git a/moto/ec2/exceptions.py b/moto/ec2/exceptions.py
index 6a93607d0..4b3373fb7 100644
--- a/moto/ec2/exceptions.py
+++ b/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):
diff --git a/moto/ec2/models/route_tables.py b/moto/ec2/models/route_tables.py
index 4215a320c..de6217d09 100644
--- a/moto/ec2/models/route_tables.py
+++ b/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,
diff --git a/moto/ec2/responses/route_tables.py b/moto/ec2/responses/route_tables.py
index bab437680..be8aeaf04 100644
--- a/moto/ec2/responses/route_tables.py
+++ b/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 = """
CreateRoute
active
{% endif %}
+ {% if route.vpc_endpoint_id %}
+ {{ route.vpc_endpoint_id }}
+ CreateRoute
+ active
+ {% endif %}
{% if route.instance %}
{{ route.instance.id }}
CreateRoute
diff --git a/tests/test_ec2/test_route_tables.py b/tests/test_ec2/test_route_tables.py
index 51c37a646..70b70bd67 100644
--- a/tests/test_ec2/test_route_tables.py
+++ b/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