146 lines
5.0 KiB
Python
146 lines
5.0 KiB
Python
from datetime import datetime
|
|
from typing import Any, Dict, List, Optional
|
|
|
|
from moto.core import CloudFormationModel
|
|
from moto.core.utils import iso_8601_datetime_with_milliseconds
|
|
from .core import TaggedEC2Resource
|
|
from ..utils import random_nat_gateway_id, random_private_ip
|
|
|
|
|
|
class NatGateway(CloudFormationModel, TaggedEC2Resource):
|
|
def __init__(
|
|
self,
|
|
backend: Any,
|
|
subnet_id: str,
|
|
allocation_id: str,
|
|
tags: Optional[Dict[str, str]] = None,
|
|
connectivity_type: str = "public",
|
|
):
|
|
# public properties
|
|
self.id = random_nat_gateway_id()
|
|
self.subnet_id = subnet_id
|
|
self.address_set: List[Dict[str, Any]] = []
|
|
self.state = "available"
|
|
self.private_ip = random_private_ip()
|
|
self.connectivity_type = connectivity_type
|
|
|
|
# protected properties
|
|
self._created_at = datetime.utcnow()
|
|
self.ec2_backend = backend
|
|
# NOTE: this is the core of NAT Gateways creation
|
|
self._eni = self.ec2_backend.create_network_interface(
|
|
backend.get_subnet(self.subnet_id), self.private_ip
|
|
)
|
|
|
|
# associate allocation with ENI
|
|
if allocation_id and connectivity_type != "private":
|
|
self.ec2_backend.associate_address(
|
|
eni=self._eni, allocation_id=allocation_id
|
|
)
|
|
self.add_tags(tags or {})
|
|
self.vpc_id = self.ec2_backend.get_subnet(subnet_id).vpc_id
|
|
|
|
@property
|
|
def physical_resource_id(self) -> str:
|
|
return self.id
|
|
|
|
@property
|
|
def create_time(self) -> str:
|
|
return iso_8601_datetime_with_milliseconds(self._created_at)
|
|
|
|
@staticmethod
|
|
def cloudformation_name_type() -> str:
|
|
return ""
|
|
|
|
@staticmethod
|
|
def cloudformation_type() -> str:
|
|
# https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-resource-ec2-natgateway.html
|
|
return "AWS::EC2::NatGateway"
|
|
|
|
@classmethod
|
|
def create_from_cloudformation_json( # type: ignore[misc]
|
|
cls,
|
|
resource_name: str,
|
|
cloudformation_json: Any,
|
|
account_id: str,
|
|
region_name: str,
|
|
**kwargs: Any
|
|
) -> "NatGateway":
|
|
from ..models import ec2_backends
|
|
|
|
ec2_backend = ec2_backends[account_id][region_name]
|
|
nat_gateway = ec2_backend.create_nat_gateway(
|
|
cloudformation_json["Properties"]["SubnetId"],
|
|
cloudformation_json["Properties"]["AllocationId"],
|
|
)
|
|
return nat_gateway
|
|
|
|
|
|
class NatGatewayBackend:
|
|
def __init__(self) -> None:
|
|
self.nat_gateways: Dict[str, NatGateway] = {}
|
|
|
|
def describe_nat_gateways(
|
|
self, filters: Any, nat_gateway_ids: Optional[List[str]]
|
|
) -> List[NatGateway]:
|
|
nat_gateways = list(self.nat_gateways.values())
|
|
|
|
if nat_gateway_ids:
|
|
nat_gateways = [item for item in nat_gateways if item.id in nat_gateway_ids]
|
|
|
|
if filters is not None:
|
|
if filters.get("nat-gateway-id") is not None:
|
|
nat_gateways = [
|
|
nat_gateway
|
|
for nat_gateway in nat_gateways
|
|
if nat_gateway.id in filters["nat-gateway-id"]
|
|
]
|
|
if filters.get("vpc-id") is not None:
|
|
nat_gateways = [
|
|
nat_gateway
|
|
for nat_gateway in nat_gateways
|
|
if nat_gateway.vpc_id in filters["vpc-id"]
|
|
]
|
|
if filters.get("subnet-id") is not None:
|
|
nat_gateways = [
|
|
nat_gateway
|
|
for nat_gateway in nat_gateways
|
|
if nat_gateway.subnet_id in filters["subnet-id"]
|
|
]
|
|
if filters.get("state") is not None:
|
|
nat_gateways = [
|
|
nat_gateway
|
|
for nat_gateway in nat_gateways
|
|
if nat_gateway.state in filters["state"]
|
|
]
|
|
|
|
return nat_gateways
|
|
|
|
def create_nat_gateway(
|
|
self,
|
|
subnet_id: str,
|
|
allocation_id: str,
|
|
tags: Optional[Dict[str, str]] = None,
|
|
connectivity_type: str = "public",
|
|
) -> NatGateway:
|
|
nat_gateway = NatGateway(
|
|
self, subnet_id, allocation_id, tags, connectivity_type
|
|
)
|
|
address_set: Dict[str, Any] = {}
|
|
if allocation_id:
|
|
eips = self.address_by_allocation([allocation_id]) # type: ignore[attr-defined]
|
|
eip = eips[0] if len(eips) > 0 else None
|
|
if eip:
|
|
address_set["allocationId"] = allocation_id
|
|
address_set["publicIp"] = eip.public_ip or None
|
|
address_set["networkInterfaceId"] = nat_gateway._eni.id
|
|
address_set["privateIp"] = nat_gateway._eni.private_ip_address
|
|
nat_gateway.address_set.append(address_set)
|
|
self.nat_gateways[nat_gateway.id] = nat_gateway
|
|
return nat_gateway
|
|
|
|
def delete_nat_gateway(self, nat_gateway_id: str) -> NatGateway:
|
|
nat_gw: NatGateway = self.nat_gateways.get(nat_gateway_id) # type: ignore
|
|
nat_gw.state = "deleted"
|
|
return nat_gw
|