msa-django-nhn/nhn/packages/vpc.py

"""
NHN Cloud VPC API Module

VPC, Subnet, Routing Table, Floating IP 관리
"""

import logging
from typing import Optional

from .base import BaseAPI, NHNCloudEndpoints, Region

logger = logging.getLogger(__name__)


class ApiVpc(BaseAPI):
    """NHN Cloud VPC API 클래스"""

    def __init__(self, region: str, token: str):
        """
        Args:
            region: 리전 (kr1: 판교, kr2: 평촌)
            token: API 인증 토큰
        """
        super().__init__(region, token)

        if self.region == Region.KR1:
            self.vpc_url = NHNCloudEndpoints.NETWORK_KR1
        else:
            self.vpc_url = NHNCloudEndpoints.NETWORK_KR2

    # ==================== VPC ====================

    def get_vpc_list(self) -> dict:
        """VPC 목록 조회"""
        url = f"{self.vpc_url}/v2.0/vpcs"
        logger.info(f"[NHN API] VPC 목록 조회 요청 - URL={url}")
        result = self._get(url)
        vpc_count = len(result.get("vpcs", []))
        logger.info(f"[NHN API] VPC 목록 조회 완료 - count={vpc_count}")
        return result

    def get_vpc_info(self, vpc_id: str) -> dict:
        """VPC 상세 조회"""
        url = f"{self.vpc_url}/v2.0/vpcs/{vpc_id}"
        logger.info(f"[NHN API] VPC 상세 조회 요청 - URL={url}, vpc_id={vpc_id}")
        result = self._get(url)
        logger.info(f"[NHN API] VPC 상세 조회 완료 - vpc_id={vpc_id}")
        return result

    def get_vpc_id_by_name(self, vpc_name: str) -> Optional[str]:
        """VPC 이름으로 ID 조회"""
        data = self.get_vpc_list()
        vpcs = data.get("vpcs", [])

        for vpc in vpcs:
            if vpc.get("name", "").startswith(vpc_name):
                return vpc.get("id")
        return None

    def create_vpc(self, name: str, cidr: str) -> dict:
        """
        VPC 생성

        Args:
            name: VPC 이름
            cidr: CIDR 블록 (예: 10.0.0.0/16)

        Returns:
            dict: 생성된 VPC 정보
        """
        url = f"{self.vpc_url}/v2.0/vpcs"
        payload = {"vpc": {"name": name, "cidrv4": cidr}}
        logger.info(f"VPC 생성 요청: {name} ({cidr})")
        return self._post(url, payload)

    def delete_vpc(self, vpc_id: str) -> dict:
        """VPC 삭제"""
        url = f"{self.vpc_url}/v2.0/vpcs/{vpc_id}"
        logger.info(f"VPC 삭제 요청: {vpc_id}")
        return self._delete(url)

    # ==================== Subnet ====================

    def get_subnet_list(self) -> dict:
        """서브넷 목록 조회"""
        url = f"{self.vpc_url}/v2.0/vpcsubnets"
        return self._get(url)

    def get_subnet_info(self, subnet_id: str) -> dict:
        """서브넷 상세 조회"""
        url = f"{self.vpc_url}/v2.0/vpcsubnets/{subnet_id}"
        return self._get(url)

    def get_subnet_id_by_name(self, subnet_name: str) -> Optional[str]:
        """서브넷 이름으로 ID 조회"""
        data = self.get_subnet_list()
        subnets = data.get("vpcsubnets", [])

        for subnet in subnets:
            name = subnet.get("name", "")
            if name.startswith(subnet_name) and "rt" not in name:
                return subnet.get("id")
        return None

    def create_subnet(self, vpc_id: str, cidr: str, name: str) -> dict:
        """
        서브넷 생성

        Args:
            vpc_id: VPC ID
            cidr: CIDR 블록 (예: 10.0.1.0/24)
            name: 서브넷 이름

        Returns:
            dict: 생성된 서브넷 정보
        """
        url = f"{self.vpc_url}/v2.0/vpcsubnets"
        payload = {"vpcsubnet": {"vpc_id": vpc_id, "cidr": cidr, "name": name}}
        logger.info(f"서브넷 생성 요청: {name} ({cidr})")
        return self._post(url, payload)

    def delete_subnet(self, subnet_id: str) -> dict:
        """서브넷 삭제"""
        url = f"{self.vpc_url}/v2.0/vpcsubnets/{subnet_id}"
        logger.info(f"서브넷 삭제 요청: {subnet_id}")
        return self._delete(url)

    # ==================== Routing Table ====================

    def get_routing_table_list(self, detail: bool = True) -> dict:
        """라우팅 테이블 목록 조회"""
        url = f"{self.vpc_url}/v2.0/routingtables"
        params = {"detail": "true"} if detail else None
        return self._get(url, params=params)

    def get_routing_table_info(self, routingtable_id: str) -> dict:
        """라우팅 테이블 상세 조회"""
        url = f"{self.vpc_url}/v2.0/routingtables/{routingtable_id}"
        return self._get(url)

    def get_default_routing_table_id(self) -> Optional[str]:
        """기본 라우팅 테이블 ID 조회"""
        data = self.get_routing_table_list(detail=False)
        tables = data.get("routingtables", [])
        return tables[0].get("id") if tables else None

    def create_routing_table(self, name: str, vpc_id: str, distributed: bool = True) -> dict:
        """
        라우팅 테이블 생성

        Args:
            name: 라우팅 테이블 이름
            vpc_id: VPC ID
            distributed: 분산 라우팅 여부 (기본 True)

        Returns:
            dict: 생성된 라우팅 테이블 정보
        """
        url = f"{self.vpc_url}/v2.0/routingtables"
        payload = {
            "routingtable": {
                "name": name,
                "vpc_id": vpc_id,
                "distributed": str(distributed).lower(),
            }
        }
        logger.info(f"라우팅 테이블 생성 요청: {name}")
        return self._post(url, payload)

    def delete_routing_table(self, routingtable_id: str) -> dict:
        """라우팅 테이블 삭제"""
        url = f"{self.vpc_url}/v2.0/routingtables/{routingtable_id}"
        logger.info(f"라우팅 테이블 삭제 요청: {routingtable_id}")
        return self._delete(url)

    def set_default_routing_table(self, routingtable_id: str) -> dict:
        """라우팅 테이블을 기본으로 설정"""
        url = f"{self.vpc_url}/v2.0/routingtables/{routingtable_id}/set_as_default"
        return self._put(url)

    def attach_gateway_to_routing_table(self, routingtable_id: str, gateway_id: str) -> dict:
        """라우팅 테이블에 인터넷 게이트웨이 연결"""
        url = f"{self.vpc_url}/v2.0/routingtables/{routingtable_id}/attach_gateway"
        payload = {"gateway_id": gateway_id}
        logger.info(f"게이트웨이 연결 요청: {routingtable_id} -> {gateway_id}")
        return self._put(url, payload)

    def detach_gateway_from_routing_table(self, routingtable_id: str) -> dict:
        """라우팅 테이블에서 인터넷 게이트웨이 분리"""
        url = f"{self.vpc_url}/v2.0/routingtables/{routingtable_id}/detach_gateway"
        return self._put(url)

    def attach_routing_table_to_subnet(self, subnet_id: str, routingtable_id: str) -> dict:
        """서브넷에 라우팅 테이블 연결"""
        url = f"{self.vpc_url}/v2.0/vpcsubnets/{subnet_id}/attach_routingtable"
        payload = {"routingtable_id": routingtable_id}
        return self._put(url, payload)

    # ==================== Internet Gateway ====================

    def get_internet_gateway_list(self) -> dict:
        """인터넷 게이트웨이 목록 조회"""
        url = f"{self.vpc_url}/v2.0/internetgateways"
        logger.info(f"[NHN API] IGW 목록 조회 요청 - URL={url}")
        result = self._get(url)
        count = len(result.get("internetgateways", []))
        logger.info(f"[NHN API] IGW 목록 조회 완료 - count={count}")
        return result

    def get_internet_gateway_info(self, internetgateway_id: str) -> dict:
        """인터넷 게이트웨이 상세 조회"""
        url = f"{self.vpc_url}/v2.0/internetgateways/{internetgateway_id}"
        return self._get(url)

    def create_internet_gateway(self, name: str, external_network_id: str) -> dict:
        """
        인터넷 게이트웨이 생성

        Args:
            name: IGW 이름
            external_network_id: 외부 네트워크 ID (get_external_network_id 로 조회)
        """
        url = f"{self.vpc_url}/v2.0/internetgateways"
        payload = {
            "internetgateway": {
                "name": name,
                "external_network_id": external_network_id,
            }
        }
        logger.info(f"IGW 생성 요청: {name} (external_network={external_network_id})")
        return self._post(url, payload)

    def delete_internet_gateway(self, internetgateway_id: str) -> dict:
        """인터넷 게이트웨이 삭제"""
        url = f"{self.vpc_url}/v2.0/internetgateways/{internetgateway_id}"
        logger.info(f"IGW 삭제 요청: {internetgateway_id}")
        return self._delete(url)

    # ==================== Floating IP ====================

    def get_external_network_id(self) -> dict:
        """외부 네트워크 ID 조회"""
        url = f"{self.vpc_url}/v2.0/networks"
        params = {"router:external": "True"}
        return self._get(url, params=params)

    def get_floating_ip_list(self) -> dict:
        """Floating IP 목록 조회"""
        url = f"{self.vpc_url}/v2.0/floatingips"
        return self._get(url)

    def create_floating_ip(self, external_network_id: str) -> dict:
        """
        Floating IP 생성

        Args:
            external_network_id: 외부 네트워크 ID

        Returns:
            dict: 생성된 Floating IP 정보
        """
        url = f"{self.vpc_url}/v2.0/floatingips"
        payload = {"floatingip": {"floating_network_id": external_network_id}}
        logger.info("Floating IP 생성 요청")
        return self._post(url, payload)

    def delete_floating_ip(self, floating_ip_id: str) -> dict:
        """Floating IP 삭제"""
        url = f"{self.vpc_url}/v2.0/floatingips/{floating_ip_id}"
        logger.info(f"Floating IP 삭제 요청: {floating_ip_id}")
        return self._delete(url)

    def attach_floating_ip(self, floating_ip_id: str, port_id: str) -> dict:
        """Floating IP를 포트에 연결"""
        url = f"{self.vpc_url}/v2.0/floatingips/{floating_ip_id}"
        payload = {"floatingip": {"port_id": port_id}}
        return self._put(url, payload)

    def detach_floating_ip(self, floating_ip_id: str) -> dict:
        """Floating IP 연결 해제"""
        url = f"{self.vpc_url}/v2.0/floatingips/{floating_ip_id}"
        payload = {"floatingip": {"port_id": None}}
        return self._put(url, payload)

    # ==================== Security Group ====================

    def get_security_group_list(self) -> dict:
        """보안 그룹 목록 조회"""
        url = f"{self.vpc_url}/v2.0/security-groups"
        return self._get(url)

    def get_security_group_info(self, security_group_id: str) -> dict:
        """보안 그룹 상세 조회"""
        url = f"{self.vpc_url}/v2.0/security-groups/{security_group_id}"
        return self._get(url)

    def create_security_group(self, name: str, description: str = "") -> dict:
        """
        보안 그룹 생성

        Args:
            name: 보안 그룹 이름
            description: 설명

        Returns:
            dict: 생성된 보안 그룹 정보
        """
        url = f"{self.vpc_url}/v2.0/security-groups"
        payload = {"security_group": {"name": name, "description": description}}
        logger.info(f"보안 그룹 생성 요청: {name}")
        return self._post(url, payload)

    def update_security_group(self, security_group_id: str, name: str = None, description: str = None) -> dict:
        """보안 그룹 수정"""
        url = f"{self.vpc_url}/v2.0/security-groups/{security_group_id}"
        payload = {"security_group": {}}
        if name is not None:
            payload["security_group"]["name"] = name
        if description is not None:
            payload["security_group"]["description"] = description
        logger.info(f"보안 그룹 수정 요청: {security_group_id}")
        return self._put(url, payload)

    def delete_security_group(self, security_group_id: str) -> dict:
        """보안 그룹 삭제"""
        url = f"{self.vpc_url}/v2.0/security-groups/{security_group_id}"
        logger.info(f"보안 그룹 삭제 요청: {security_group_id}")
        return self._delete(url)

    # ==================== Security Group Rule ====================

    def get_security_group_rule_list(self, security_group_id: str = None) -> dict:
        """보안 그룹 규칙 목록 조회"""
        url = f"{self.vpc_url}/v2.0/security-group-rules"
        params = {}
        if security_group_id:
            params["security_group_id"] = security_group_id
        return self._get(url, params=params if params else None)

    def create_security_group_rule(
        self,
        security_group_id: str,
        direction: str,
        ethertype: str = "IPv4",
        protocol: str = None,
        port_range_min: int = None,
        port_range_max: int = None,
        remote_ip_prefix: str = None,
        remote_group_id: str = None,
        description: str = "",
    ) -> dict:
        """
        보안 그룹 규칙 생성

        Args:
            security_group_id: 보안 그룹 ID
            direction: 방향 (ingress/egress)
            ethertype: IPv4/IPv6
            protocol: 프로토콜 (tcp/udp/icmp 등)
            port_range_min: 최소 포트
            port_range_max: 최대 포트
            remote_ip_prefix: 원격 IP 대역 (CIDR)
            remote_group_id: 원격 보안 그룹 ID
            description: 설명
        """
        url = f"{self.vpc_url}/v2.0/security-group-rules"
        rule = {
            "security_group_id": security_group_id,
            "direction": direction,
            "ethertype": ethertype,
        }
        if protocol:
            rule["protocol"] = protocol
        if port_range_min is not None:
            rule["port_range_min"] = port_range_min
        if port_range_max is not None:
            rule["port_range_max"] = port_range_max
        if remote_ip_prefix:
            rule["remote_ip_prefix"] = remote_ip_prefix
        if remote_group_id:
            rule["remote_group_id"] = remote_group_id
        if description:
            rule["description"] = description
        payload = {"security_group_rule": rule}
        logger.info(f"보안 그룹 규칙 생성 요청: sg={security_group_id}, dir={direction}, proto={protocol}")
        return self._post(url, payload)

    def delete_security_group_rule(self, rule_id: str) -> dict:
        """보안 그룹 규칙 삭제"""
        url = f"{self.vpc_url}/v2.0/security-group-rules/{rule_id}"
        logger.info(f"보안 그룹 규칙 삭제 요청: {rule_id}")
        return self._delete(url)

    # ==================== Port (NIC) ====================

    def get_port_list(self) -> dict:
        """포트 목록 조회"""
        url = f"{self.vpc_url}/v2.0/ports"
        return self._get(url)

    def get_port_id_by_device(self, device_id: str) -> Optional[str]:
        """디바이스 ID로 포트 ID 조회"""
        data = self.get_port_list()
        ports = data.get("ports", [])

        for port in ports:
            if port.get("device_id", "").startswith(device_id):
                return port.get("id")
        return None