105 lines
2.9 KiB
Python
105 lines
2.9 KiB
Python
"""AWS EC2 Elastic IP 操作"""
|
|
from __future__ import annotations
|
|
|
|
import re
|
|
from dataclasses import dataclass
|
|
from typing import Any
|
|
|
|
import boto3
|
|
|
|
from .aws_region import normalize_aws_region
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class ElasticIpInfo:
|
|
allocation_id: str
|
|
association_id: str | None
|
|
public_ip: str | None
|
|
|
|
|
|
_INSTANCE_ID_RE = re.compile(r"^i-[0-9a-f]{8,17}$", re.IGNORECASE)
|
|
|
|
|
|
def is_valid_instance_id(value: str | None) -> bool:
|
|
if not value:
|
|
return False
|
|
return bool(_INSTANCE_ID_RE.match(value.strip()))
|
|
|
|
|
|
def create_ec2_client(
|
|
*,
|
|
region: str,
|
|
aws_access_key: str | None,
|
|
aws_secret_key: str | None,
|
|
):
|
|
kwargs: dict[str, Any] = {"region_name": normalize_aws_region(region)}
|
|
if aws_access_key and aws_secret_key:
|
|
kwargs["aws_access_key_id"] = aws_access_key
|
|
kwargs["aws_secret_access_key"] = aws_secret_key
|
|
return boto3.client("ec2", **kwargs)
|
|
|
|
|
|
def get_instance_elastic_ip(ec2_client, *, instance_id: str) -> ElasticIpInfo | None:
|
|
resp = ec2_client.describe_addresses(
|
|
Filters=[{"Name": "instance-id", "Values": [instance_id]}]
|
|
)
|
|
addresses = resp.get("Addresses") or []
|
|
if not addresses:
|
|
return None
|
|
|
|
addr = addresses[0]
|
|
allocation_id = addr.get("AllocationId")
|
|
if not allocation_id:
|
|
return None
|
|
|
|
return ElasticIpInfo(
|
|
allocation_id=allocation_id,
|
|
association_id=addr.get("AssociationId"),
|
|
public_ip=addr.get("PublicIp"),
|
|
)
|
|
|
|
|
|
def disassociate_elastic_ip(ec2_client, *, association_id: str) -> None:
|
|
ec2_client.disassociate_address(AssociationId=association_id)
|
|
|
|
|
|
def release_elastic_ip(ec2_client, *, allocation_id: str) -> None:
|
|
ec2_client.release_address(AllocationId=allocation_id)
|
|
|
|
|
|
def allocate_elastic_ip(ec2_client) -> ElasticIpInfo:
|
|
resp = ec2_client.allocate_address(Domain="vpc")
|
|
return ElasticIpInfo(
|
|
allocation_id=resp["AllocationId"],
|
|
association_id=None,
|
|
public_ip=resp.get("PublicIp"),
|
|
)
|
|
|
|
|
|
def associate_elastic_ip(ec2_client, *, instance_id: str, allocation_id: str) -> str:
|
|
resp = ec2_client.associate_address(InstanceId=instance_id, AllocationId=allocation_id)
|
|
return resp.get("AssociationId") or ""
|
|
|
|
|
|
def rotate_elastic_ip(
|
|
ec2_client,
|
|
*,
|
|
instance_id: str,
|
|
release_old: bool,
|
|
) -> dict[str, str | None]:
|
|
current = get_instance_elastic_ip(ec2_client, instance_id=instance_id)
|
|
|
|
if current and current.association_id:
|
|
disassociate_elastic_ip(ec2_client, association_id=current.association_id)
|
|
|
|
if current and release_old:
|
|
release_elastic_ip(ec2_client, allocation_id=current.allocation_id)
|
|
|
|
new_eip = allocate_elastic_ip(ec2_client)
|
|
associate_elastic_ip(ec2_client, instance_id=instance_id, allocation_id=new_eip.allocation_id)
|
|
|
|
return {
|
|
"public_ip": new_eip.public_ip,
|
|
"allocation_id": new_eip.allocation_id,
|
|
}
|