feat:自动更换ip+流量监控

services/aws_eip.py

@@ -0,0 +1,104 @@
+"""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,
+    }