fix: 전반적인 코드 개선 3차

Author: MU-Software

app/admin_api/test/cms_test.py

@@ -9,12 +9,15 @@
 
 @pytest.fixture
 def domain_group(superuser):
-    return DomainGroup.objects.create(
+    obj, _ = DomainGroup.objects.get_or_create(
         name="2025년 PyConKR 홈페이지",
-        domains=["2025.pycon.kr"],
-        created_by=superuser,
-        updated_by=superuser,
+        defaults={
+            "domains": ["2025.pycon.kr"],
+            "created_by": superuser,
+            "updated_by": superuser,
+        },
     )
+    return obj
 
 
 # ---- Auth -------------------------------------------------------------------

app/admin_api/views/modification_audit.py

@@ -25,10 +25,7 @@
 }
 
 
-@utils.extend_schema_view(
-    list=utils.extend_schema(tags=[OpenAPITag.ADMIN_MODIFICATION_AUDIT]),
-    retrieve=utils.extend_schema(tags=[OpenAPITag.ADMIN_MODIFICATION_AUDIT]),
-)
+@utils.extend_schema_view(list=utils.extend_schema(tags=[OpenAPITag.ADMIN_MODIFICATION_AUDIT]))
 class ModificationAuditAdminViewSet(mixins.ListModelMixin, viewsets.GenericViewSet):
     serializer_class = ModificationAuditResponseAdminSerializer
     permission_classes = [IsSuperUser]

app/admin_api/views/user.py

@@ -15,10 +15,11 @@
 from user.models import UserExt
 from user.models.organization import Organization
 
-ADMIN_METHODS = ["list", "retrieve", "create", "partial_update", "destroy"]
+USER_ADMIN_METHODS = ["list", "retrieve", "create", "partial_update"]
+ADMIN_METHODS = USER_ADMIN_METHODS + ["destroy"]
 
 
-@extend_schema_view(**{m: extend_schema(tags=[OpenAPITag.ADMIN_USER]) for m in ADMIN_METHODS})
+@extend_schema_view(**{m: extend_schema(tags=[OpenAPITag.ADMIN_USER]) for m in USER_ADMIN_METHODS})
 class UserAdminViewSet(
     mixins.RetrieveModelMixin,
     mixins.ListModelMixin,

app/cms/test/sitemap_api_test.py

@@ -24,7 +24,7 @@ def test_list_view(api_client, create_sitemap):
 @pytest.mark.django_db
 def test_list_view_returns_only_matching_domain(api_client):
     group_main = DomainGroup.objects.create(name="main", domains=["pycon.kr"])
-    group_legacy = DomainGroup.objects.create(name="legacy", domains=["2025.pycon.kr"])
+    group_legacy = DomainGroup.objects.create(name="legacy", domains=["legacy.pycon.kr"])
 
     _create_sitemap("main_about", group=group_main)
     _create_sitemap("legacy_about", group=group_legacy)

app/core/authn/api_key.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+from hmac import compare_digest
 from typing import TYPE_CHECKING
 
 from django.conf import settings
@@ -30,10 +31,11 @@ def authenticate(self, request: Request) -> tuple["UserExt", None] | None:
         api_key = request.headers.get("x-api-key", "")
         api_secret = request.headers.get("x-api-secret", "")
 
-        if api_key.lower() in settings.EXT_API_KEYS and api_secret == settings.EXT_API_KEYS.get(api_key.lower()):
-            return self._get_or_create_api_key_user(api_key), None
-
-        return None
+        if not (expected_secret := settings.EXT_API_KEYS.get(api_key.lower())):
+            return None
+        if not compare_digest(api_secret.encode(), expected_secret.encode()):
+            return None
+        return self._get_or_create_api_key_user(api_key), None
 
 
 class APIKeyAuthenticationScheme(OpenApiAuthenticationExtension):  # type: ignore[no-untyped-call]

app/core/external_apis/portone/client.py

@@ -1,7 +1,8 @@
 from __future__ import annotations
 
-from datetime import datetime
+from datetime import datetime, timedelta
 from logging import getLogger
+from time import time
 from traceback import format_exception
 from typing import Any, Literal
 
@@ -14,6 +15,11 @@
 logger = getLogger(__name__)
 
 DEFAULT_TIMEOUT = 5
+# PortOne v1 access_token 의 공식 TTL 은 발행 시점 +30분 (developers.portone.io 명시).
+# 응답에 `expired_at` (unix epoch sec) 가 포함되며, 만료 직전 재발급 race 를 막기 위한 안전 마진.
+TOKEN_REFRESH_MARGIN = timedelta(seconds=30)
+# `expired_at` 가 응답에 누락된 비정상 케이스의 보수적 fallback (공식 TTL 30분보다 짧게).
+TOKEN_FALLBACK_TTL = timedelta(minutes=5)
 RequestMethodType = Literal["GET", "OPTIONS", "HEAD", "POST", "PUT", "PATCH", "DELETE"]
 
 
@@ -29,6 +35,8 @@ class PortOneClient:
     def __init__(self, timeout: TimeoutTypes = DEFAULT_TIMEOUT) -> None:
         self._timeout = timeout
         self._client: Client | None = None
+        self._cached_token: str | None = None
+        self._cached_token_expires_at: float = 0.0
 
     @property
     def client(self) -> Client:
@@ -39,6 +47,10 @@ def client(self) -> Client:
 
     @property
     def _access_token(self) -> str:
+        # 만료 직전 안전 마진까지는 캐시 재사용 (PortOne 토큰 TTL 30분).
+        if self._cached_token and time() < self._cached_token_expires_at - TOKEN_REFRESH_MARGIN.total_seconds():
+            return self._cached_token
+
         response = self.client.post(
             url="/users/getToken", json={"imp_key": settings.PORTONE.imp_key, "imp_secret": settings.PORTONE.imp_secret}
         )
@@ -47,15 +59,18 @@ def _access_token(self) -> str:
             resp_serializer = PortOneV1ResponseSerializer.from_response(response)
             resp_serializer.is_valid(raise_exception=True)
 
-            if not (access_token := resp_serializer.validated_data["response"]["access_token"]):
+            resp = resp_serializer.validated_data["response"]
+            if not (access_token := resp.get("access_token")):
                 raise ValueError("PortOne access_token 값이 존재하지 않습니다.")
 
-            return access_token
-
         except Exception as e:
             logger.error(format_exception(e))
             raise PortOneException("PortOne AccessToken 획득에 실패했습니다.") from e
 
+        self._cached_token = access_token
+        self._cached_token_expires_at = float(resp.get("expired_at") or (time() + TOKEN_FALLBACK_TTL.total_seconds()))
+        return access_token
+
     def _request(
         self,
         method: RequestMethodType,

app/core/util/strutil.py

@@ -14,5 +14,4 @@ def uuid_to_b64(in_str: UUID | str) -> str:
 
 
 def b64_to_uuid(in_str: str) -> UUID:
-    in_str += "=" * (4 - len(in_str) % 4)
-    return UUID(bytes=urlsafe_b64decode(in_str + "=="))
+    return UUID(bytes=urlsafe_b64decode(in_str + "=" * (-len(in_str) % 4)))

app/core/util/totp.py

@@ -1,3 +1,4 @@
+import warnings
 from base64 import b32encode
 from dataclasses import dataclass
 from hmac import new as hmac_new
@@ -21,9 +22,15 @@ def __post_init__(self) -> None:
         if self.digest not in ALLOWED_DIGESTS:
             raise ValueError(f"Unsupported digest algorithm: {self.digest}")
         if self.digest != "sha1":
-            raise Warning(f"Using {self.digest} is not recommended as Google Authenticator does not support it.")
+            warnings.warn(
+                f"Using {self.digest} is not recommended as Google Authenticator does not support it.",
+                stacklevel=2,
+            )
         if self.time_step != 30:
-            raise Warning(f"Using {self.time_step} is not recommended as Google Authenticator does not support it.")
+            warnings.warn(
+                f"Using time_step={self.time_step} is not recommended as Google Authenticator does not support it.",
+                stacklevel=2,
+            )
 
     def get_hotp(self, counter: int) -> str:
         mac = hmac_new(self.key, pack(">Q", counter), self.digest).digest()

app/internal_api/views.py

@@ -89,10 +89,11 @@ class DeskSupportViewSet(
         parameters=[
             OpenApiParameter(
                 name="otp",
-                type=OpenApiTypes.INT,
+                type=OpenApiTypes.STR,
                 location=OpenApiParameter.QUERY,
                 allow_blank=False,
                 required=True,
+                description="환불 승인자의 6자리 TOTP 코드",
             ),
         ],
         responses={
@@ -112,7 +113,7 @@ def destroy(
         """
         serializer = OrderTotalRefundSerializer(
             instance=self.get_object(),
-            data={"totp": request.GET.get("otp")},
+            data={"totp": request.query_params.get("otp")},
             context={"check_refundable_date": False},
         )
         serializer.is_valid(raise_exception=True)

app/shop/order/views/orders.py

@@ -270,9 +270,9 @@ def retrieve_receipt(
 
 class OrderProductViewSet(mixins.DestroyModelMixin, viewsets.GenericViewSet):
     lookup_url_kwarg = "order_product_rel_id"
-    serializer_class = OrderDto
+    serializer_class = None
 
-    def get_queryset(self) -> models.QuerySet[Order]:
+    def get_queryset(self) -> models.QuerySet[OrderProductRelation]:
         if not isinstance(self.request.user, UserExt):
             return OrderProductRelation.objects.none()