v0.0.17 | 보안그룹 관리 API 추가 (CRUD + 규칙 CRUD)
All checks were successful
Build And Test / build-and-push (push) Successful in 53s

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
2026-03-18 13:57:08 +09:00
parent 9bf41ebf21
commit 2eade2ee9b
5 changed files with 332 additions and 1 deletions

View File

@ -254,6 +254,103 @@ class ApiVpc(BaseAPI):
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:

View File

@ -119,6 +119,92 @@ class SubnetSerializer(serializers.Serializer):
)
# ==================== Security Group ====================
class SecurityGroupSerializer(serializers.Serializer):
"""보안 그룹 생성 요청"""
name = serializers.CharField(
help_text="보안 그룹 이름",
max_length=255,
)
description = serializers.CharField(
help_text="설명",
required=False,
allow_blank=True,
default="",
)
class SecurityGroupUpdateSerializer(serializers.Serializer):
"""보안 그룹 수정 요청"""
name = serializers.CharField(
help_text="보안 그룹 이름",
required=False,
max_length=255,
)
description = serializers.CharField(
help_text="설명",
required=False,
allow_blank=True,
)
class SecurityGroupRuleSerializer(serializers.Serializer):
"""보안 그룹 규칙 생성 요청"""
security_group_id = serializers.CharField(
help_text="보안 그룹 ID",
)
direction = serializers.ChoiceField(
choices=["ingress", "egress"],
help_text="방향 (ingress: 인바운드, egress: 아웃바운드)",
)
ethertype = serializers.ChoiceField(
choices=["IPv4", "IPv6"],
help_text="IP 버전",
default="IPv4",
)
protocol = serializers.ChoiceField(
choices=["tcp", "udp", "icmp", ""],
help_text="프로토콜 (tcp, udp, icmp, 빈값=전체)",
required=False,
allow_blank=True,
)
port_range_min = serializers.IntegerField(
help_text="최소 포트 번호",
required=False,
allow_null=True,
min_value=1,
max_value=65535,
)
port_range_max = serializers.IntegerField(
help_text="최대 포트 번호",
required=False,
allow_null=True,
min_value=1,
max_value=65535,
)
remote_ip_prefix = serializers.CharField(
help_text="원격 IP 대역 (CIDR, 예: 0.0.0.0/0)",
required=False,
allow_blank=True,
)
remote_group_id = serializers.CharField(
help_text="원격 보안 그룹 ID",
required=False,
allow_blank=True,
)
description = serializers.CharField(
help_text="설명",
required=False,
allow_blank=True,
default="",
)
# ==================== NKS ====================

View File

@ -36,6 +36,12 @@ urlpatterns = [
# ==================== Security Group ====================
path("securitygroup/", views.SecurityGroupListView.as_view(), name="securitygroup-list"),
path("securitygroup/create/", views.SecurityGroupCreateView.as_view(), name="securitygroup-create"),
path("securitygroup/<str:security_group_id>/", views.SecurityGroupDetailView.as_view(), name="securitygroup-detail"),
# ==================== Security Group Rule ====================
path("securitygroup-rule/create/", views.SecurityGroupRuleCreateView.as_view(), name="securitygroup-rule-create"),
path("securitygroup-rule/<str:rule_id>/", views.SecurityGroupRuleDeleteView.as_view(), name="securitygroup-rule-delete"),
# ==================== Async Task ====================
path("tasks/", views.AsyncTaskListView.as_view(), name="task-list"),

View File

@ -19,6 +19,9 @@ from .serializers import (
ComputeInstanceSerializer,
VpcSerializer,
SubnetSerializer,
SecurityGroupSerializer,
SecurityGroupUpdateSerializer,
SecurityGroupRuleSerializer,
NksClusterSerializer,
NksNodeGroupSerializer,
NksNodeActionSerializer,
@ -671,6 +674,145 @@ class SecurityGroupListView(NHNBaseView):
return Response({"error": e.message}, status=status.HTTP_400_BAD_REQUEST)
class SecurityGroupCreateView(NHNBaseView):
"""보안 그룹 생성 API"""
@swagger_auto_schema(
operation_summary="보안 그룹 생성",
manual_parameters=[region_header, token_header],
request_body=SecurityGroupSerializer,
responses={201: "생성된 보안 그룹 정보"},
)
def post(self, request):
headers = get_nhn_headers(request)
serializer = SecurityGroupSerializer(data=request.data)
if not serializer.is_valid():
return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST)
try:
api = ApiVpc(headers["region"], headers["token"])
result = api.create_security_group(
name=serializer.validated_data["name"],
description=serializer.validated_data.get("description", ""),
)
logger.info(f"[SG] 보안 그룹 생성 성공: {serializer.validated_data['name']}")
return Response(result, status=status.HTTP_201_CREATED)
except NHNCloudAPIError as e:
logger.error(f"[SG] 보안 그룹 생성 실패 - error={e.message}")
return Response({"error": e.message}, status=status.HTTP_400_BAD_REQUEST)
class SecurityGroupDetailView(NHNBaseView):
"""보안 그룹 상세 조회/수정/삭제 API"""
@swagger_auto_schema(
operation_summary="보안 그룹 상세 조회",
manual_parameters=[region_header, token_header],
responses={200: "보안 그룹 상세 정보"},
)
def get(self, request, security_group_id):
headers = get_nhn_headers(request)
try:
api = ApiVpc(headers["region"], headers["token"])
return Response(api.get_security_group_info(security_group_id))
except NHNCloudAPIError as e:
return Response({"error": e.message}, status=status.HTTP_400_BAD_REQUEST)
@swagger_auto_schema(
operation_summary="보안 그룹 수정",
manual_parameters=[region_header, token_header],
request_body=SecurityGroupUpdateSerializer,
responses={200: "수정된 보안 그룹 정보"},
)
def put(self, request, security_group_id):
headers = get_nhn_headers(request)
serializer = SecurityGroupUpdateSerializer(data=request.data)
if not serializer.is_valid():
return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST)
try:
api = ApiVpc(headers["region"], headers["token"])
result = api.update_security_group(
security_group_id,
name=serializer.validated_data.get("name"),
description=serializer.validated_data.get("description"),
)
logger.info(f"[SG] 보안 그룹 수정 성공: {security_group_id}")
return Response(result)
except NHNCloudAPIError as e:
logger.error(f"[SG] 보안 그룹 수정 실패 - error={e.message}")
return Response({"error": e.message}, status=status.HTTP_400_BAD_REQUEST)
@swagger_auto_schema(
operation_summary="보안 그룹 삭제",
manual_parameters=[region_header, token_header],
responses={204: "삭제 성공"},
)
def delete(self, request, security_group_id):
headers = get_nhn_headers(request)
try:
api = ApiVpc(headers["region"], headers["token"])
api.delete_security_group(security_group_id)
logger.info(f"[SG] 보안 그룹 삭제 성공: {security_group_id}")
return Response(status=status.HTTP_204_NO_CONTENT)
except NHNCloudAPIError as e:
logger.error(f"[SG] 보안 그룹 삭제 실패 - error={e.message}")
return Response({"error": e.message}, status=status.HTTP_400_BAD_REQUEST)
class SecurityGroupRuleCreateView(NHNBaseView):
"""보안 그룹 규칙 생성 API"""
@swagger_auto_schema(
operation_summary="보안 그룹 규칙 생성",
manual_parameters=[region_header, token_header],
request_body=SecurityGroupRuleSerializer,
responses={201: "생성된 규칙 정보"},
)
def post(self, request):
headers = get_nhn_headers(request)
serializer = SecurityGroupRuleSerializer(data=request.data)
if not serializer.is_valid():
return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST)
try:
data = serializer.validated_data
api = ApiVpc(headers["region"], headers["token"])
result = api.create_security_group_rule(
security_group_id=data["security_group_id"],
direction=data["direction"],
ethertype=data.get("ethertype", "IPv4"),
protocol=data.get("protocol") or None,
port_range_min=data.get("port_range_min"),
port_range_max=data.get("port_range_max"),
remote_ip_prefix=data.get("remote_ip_prefix") or None,
remote_group_id=data.get("remote_group_id") or None,
description=data.get("description", ""),
)
logger.info(f"[SG Rule] 규칙 생성 성공: sg={data['security_group_id']}")
return Response(result, status=status.HTTP_201_CREATED)
except NHNCloudAPIError as e:
logger.error(f"[SG Rule] 규칙 생성 실패 - error={e.message}")
return Response({"error": e.message}, status=status.HTTP_400_BAD_REQUEST)
class SecurityGroupRuleDeleteView(NHNBaseView):
"""보안 그룹 규칙 삭제 API"""
@swagger_auto_schema(
operation_summary="보안 그룹 규칙 삭제",
manual_parameters=[region_header, token_header],
responses={204: "삭제 성공"},
)
def delete(self, request, rule_id):
headers = get_nhn_headers(request)
try:
api = ApiVpc(headers["region"], headers["token"])
api.delete_security_group_rule(rule_id)
logger.info(f"[SG Rule] 규칙 삭제 성공: {rule_id}")
return Response(status=status.HTTP_204_NO_CONTENT)
except NHNCloudAPIError as e:
logger.error(f"[SG Rule] 규칙 삭제 실패 - error={e.message}")
return Response({"error": e.message}, status=status.HTTP_400_BAD_REQUEST)
# ==================== Async Task API ====================