security: 결제 진행 중 장바구니 변경을 막음

Author: MU-Software

app/core/const/shop_error_messages.py

@@ -1,3 +1,7 @@
+from django.db import models
+from rest_framework.exceptions import ValidationError
+
+
 class CriticalErrorMessages:
     INVALID_LOGIC = "발생하면 안 되는 오류가 발생했습니다. PyCon 한국 준비 위원회에 문의해주세요.\n{}"
 
@@ -80,13 +84,19 @@ class OptionGroupNotModifiableErrorMessages:
     RESPONSE_NOT_MODIFIABLE = "해당 옵션은 수정할 수 없습니다. PyCon 한국 준비 위원회에 문의해주세요."
 
 
-class PortOneWebhookFailureMessages:
-    ORDER_NOT_FOUND = "주문 정보가 존재하지 않습니다."
-    PURCHASE_FAILED = "결제에 실패했습니다."
-    VIRTUAL_ACCOUNT_NOT_SUPPORTED = "가상계좌 결제는 지원하지 않습니다."
-    UNEXPECTED_RETRIEVED_ORDER_STATUS = "예상한 결제 상태가 아닙니다."
-    UNEXPECTED_RETRIEVED_ORDER_ID = "결제 ID가 일치하지 않습니다."
-    UNEXPECTED_PAID_PRICE = "결제 금액이 일치하지 않습니다."
-    UNSUPPORTED_CURRENCY = "지원하지 않는 통화입니다."
-    ILLEGAL_STATUS_TRANSITION = "이미 처리된 결제이거나 허용되지 않는 상태 전환입니다."
-    CANCELLED_NOT_SUPPORTED = "관리자 콘솔에서 결제 취소된 webhook 의 자동 처리는 아직 지원하지 않습니다."
+class PortOneWebhookFailureCode(models.TextChoices):
+    ORDER_NOT_FOUND = "ORDER_NOT_FOUND", "주문 정보가 존재하지 않습니다."
+    PURCHASE_FAILED = "PURCHASE_FAILED", "결제에 실패했습니다."
+    VIRTUAL_ACCOUNT_NOT_SUPPORTED = "VIRTUAL_ACCOUNT_NOT_SUPPORTED", "가상계좌 결제는 지원하지 않습니다."
+    UNEXPECTED_RETRIEVED_ORDER_STATUS = "UNEXPECTED_RETRIEVED_ORDER_STATUS", "예상한 결제 상태가 아닙니다."
+    UNEXPECTED_RETRIEVED_ORDER_ID = "UNEXPECTED_RETRIEVED_ORDER_ID", "결제 ID가 일치하지 않습니다."
+    UNEXPECTED_PAID_PRICE = "UNEXPECTED_PAID_PRICE", "결제 금액이 일치하지 않습니다."
+    UNSUPPORTED_CURRENCY = "UNSUPPORTED_CURRENCY", "지원하지 않는 통화입니다."
+    ILLEGAL_STATUS_TRANSITION = "ILLEGAL_STATUS_TRANSITION", "이미 처리된 결제이거나 허용되지 않는 상태 전환입니다."
+    CANCELLED_NOT_SUPPORTED = (
+        "CANCELLED_NOT_SUPPORTED",
+        "관리자 콘솔에서 결제 취소된 webhook 의 자동 처리는 아직 지원하지 않습니다.",
+    )
+
+    def as_error(self) -> ValidationError:
+        return ValidationError(detail=self.label, code=self.value)

app/core/external_apis/portone/client.py

@@ -149,19 +149,23 @@ def find_payment_info(self, imp_uid: str) -> dict:
         raise PortOneException(f"결제 정보를 찾을 수 없습니다. {imp_uid=}")
 
     def req_cancel_payment(
-        self, merchant_id: str, refund_request_price: int, current_leftover_price: int, reason: str | None = None
+        self,
+        imp_id: str,
+        refund_request_price: int | float,
+        current_leftover_price: int | float,
+        reason: str | None = None,
     ) -> dict:
         """결제 환불 요청
         Args:
-            merchant_id (str): 결제 번호
-            refund_request_price (int): 환불 요청 금액
-            current_leftover_price (int): 현재 환불 가능한 남은 금액
+            imp_id (str): 포트원 거래고유번호
+            refund_request_price (int | float): 환불 요청 금액
+            current_leftover_price (int | float): 현재 환불 가능한 남은 금액
             reason (str | None): 환불 사유
         Returns:
             dict: 결제 사전 등록 또는 수정 응답
         """
         request_dto = {
-            "merchant_uid": merchant_id,
+            "imp_uid": imp_id,
             "amount": refund_request_price,
             "checksum": current_leftover_price,
             "reason": reason,

app/shop/conftest.py

@@ -197,7 +197,7 @@ def single_product_cart(customer_user, product) -> SingleProductCart:
     return cart
 
 
-OrderStatus = Literal["empty", "cart", "completed", "refunded", "partial_refunded"]
+OrderStatus = Literal["empty", "cart", "prepared", "completed", "refunded", "partial_refunded"]
 
 
 @pytest.fixture
@@ -208,6 +208,7 @@ def order_factory(customer_user, product, donation_product):
         status:
             - ``"empty"``: OPR / CustomerInfo 없는 빈 Order (donation 인자 무시)
             - ``"cart"``: PH 없음, OPR pending
+            - ``"prepared"``: ``"cart"`` + 결제 준비 snapshot/hash 저장
             - ``"completed"``: PH completed + OPR paid
             - ``"refunded"``: 전액 환불 — PH refunded(price=0) 추가 + OPR refunded
             - ``"partial_refunded"``: 부분 환불 — PH partial_refunded 추가
@@ -232,6 +233,9 @@ def make(*, status: OrderStatus = "cart", donation: int = 0) -> Order:
 
         if status == "cart":
             return order
+        if status == "prepared":
+            order.prepare_payment()
+            return order
 
         # status >= 'completed' — OPR paid + PH completed.
         order.products.update(status=OrderProductRelation.OrderProductStatus.paid)
@@ -304,6 +308,7 @@ def mock_portone_req_cancel_payment():
     환불 호출은 반환값을 호출자가 사용하지 않으므로 (호출 인자/횟수만 검증), 명시 세팅 없이도 None 반환 허용.
     """
     with patch.object(portone_client, "req_cancel_payment") as mocked:
+        mocked.return_value = None
         yield mocked
 
 

app/shop/order/migrations/0003_order_payment_preparation.py

@@ -0,0 +1,49 @@
+# Generated by Django 6.0.4 on 2026-05-23 11:52
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+    dependencies = [("order", "0002_migrate_from_legacy")]
+    operations = [
+        migrations.AddField(
+            model_name="historicalorder",
+            name="prepared_cart_hash",
+            field=models.CharField(blank=True, max_length=16, null=True),
+        ),
+        migrations.AddField(
+            model_name="historicalorder",
+            name="prepared_cart_snapshot",
+            field=models.JSONField(blank=True, null=True),
+        ),
+        migrations.AddField(
+            model_name="historicalsingleproductcart",
+            name="prepared_cart_hash",
+            field=models.CharField(blank=True, max_length=16, null=True),
+        ),
+        migrations.AddField(
+            model_name="historicalsingleproductcart",
+            name="prepared_cart_snapshot",
+            field=models.JSONField(blank=True, null=True),
+        ),
+        migrations.AddField(
+            model_name="order",
+            name="prepared_cart_hash",
+            field=models.CharField(blank=True, max_length=16, null=True),
+        ),
+        migrations.AddField(
+            model_name="order",
+            name="prepared_cart_snapshot",
+            field=models.JSONField(blank=True, null=True),
+        ),
+        migrations.AddField(
+            model_name="singleproductcart",
+            name="prepared_cart_hash",
+            field=models.CharField(blank=True, max_length=16, null=True),
+        ),
+        migrations.AddField(
+            model_name="singleproductcart",
+            name="prepared_cart_snapshot",
+            field=models.JSONField(blank=True, null=True),
+        ),
+    ]

app/shop/order/models.py

@@ -2,9 +2,17 @@
 
 import datetime
 import functools
+import json
 import typing
+from base64 import urlsafe_b64encode
+from collections.abc import Iterable
+from contextlib import suppress
+from hashlib import sha256
+from hmac import new as hmac_new
 from urllib.parse import urljoin
+from uuid import UUID, uuid4
 
+import shortuuid
 from core.const.shop_error_messages import NotRefundableErrorMessages
 from core.models import BaseAbstractModel, BaseAbstractModelQuerySet
 from core.scancode_mixin import ScanCodeMixin
@@ -17,9 +25,97 @@
 from simple_history.models import HistoricalRecords
 
 UserModel = get_user_model()
+PAYMENT_HASH_LENGTH = 16
 
 
-class OrderQuerySet(BaseAbstractModelQuerySet):
+class PaymentPreparationMixin:
+    id: UUID
+    prepared_cart_snapshot: dict[str, typing.Any] | None
+    prepared_cart_hash: str | None
+    first_paid_price: int
+    products: typing.Any
+
+    @property
+    def merchant_uid(self) -> str | None:
+        return f"{shortuuid.encode(self.id)}.{self.prepared_cart_hash}" if self.prepared_cart_hash else None
+
+    @property
+    def prepared_price(self) -> int | None:
+        return self.prepared_cart_snapshot["price"] if self.prepared_cart_snapshot else None
+
+    def get_current_cart_snapshot(self, *, attempt: dict[str, str] | None = None) -> dict[str, typing.Any]:
+        products = self.products.filter_active().prefetch_related(
+            models.Prefetch("options", queryset=OrderProductOptionRelation.objects.filter_active()),
+        )
+        return {
+            "attempt": attempt or {"id": str(uuid4()), "prepared_at": now_aware().isoformat()},
+            "price": self.first_paid_price,
+            "products": [
+                {
+                    "id": str(product_rel.id),
+                    "product": str(product_rel.product_id),
+                    "price": product_rel.price,
+                    "donation_price": product_rel.donation_price,
+                    "options": [
+                        {
+                            "id": str(option_rel.id),
+                            "product_option_group": str(option_rel.product_option_group_id),
+                            "product_option": (
+                                str(option_rel.product_option_id) if option_rel.product_option_id is not None else None
+                            ),
+                            "custom_response": option_rel.custom_response,
+                        }
+                        for option_rel in sorted(product_rel.options.all(), key=lambda option_rel: str(option_rel.id))
+                    ],
+                }
+                for product_rel in sorted(products, key=lambda rel: str(rel.id))
+            ],
+        }
+
+    @staticmethod
+    def _compute_cart_hash(snapshot: dict[str, typing.Any]) -> str:
+        digest = hmac_new(
+            settings.SECRET_KEY.encode(),
+            json.dumps(snapshot, ensure_ascii=True, separators=(",", ":"), sort_keys=True).encode(),
+            sha256,
+        ).digest()
+        return urlsafe_b64encode(digest[:12]).decode().rstrip("=")
+
+    def get_current_cart_hash(self, *, attempt: dict[str, str] | None = None) -> str:
+        return self._compute_cart_hash(self.get_current_cart_snapshot(attempt=attempt))
+
+    def prepare_payment(self) -> None:
+        snapshot = self.get_current_cart_snapshot()
+        self.prepared_cart_snapshot = snapshot
+        self.prepared_cart_hash = self._compute_cart_hash(snapshot)
+        self.save(update_fields={"prepared_cart_snapshot", "prepared_cart_hash"})
+
+    def matches_payment_preparation(self, merchant_uid: str, amount: int | float) -> bool:
+        if not self.prepared_cart_snapshot:
+            return False
+        try:
+            amount_int = int(amount)
+        except (TypeError, ValueError):
+            return False
+        attempt = self.prepared_cart_snapshot.get("attempt")
+        if not isinstance(attempt, dict):
+            return False
+        return (
+            self.merchant_uid == merchant_uid
+            and amount == self.prepared_price == amount_int
+            and self.prepared_cart_hash == self.get_current_cart_hash(attempt=attempt)
+        )
+
+
+class BaseCartQuerySet(BaseAbstractModelQuerySet):
+    def filter_by_merchant_uid(self, merchant_uid: object) -> models.QuerySet:
+        with suppress(AttributeError, TypeError, ValueError):
+            encoded_id, cart_hash = merchant_uid.split(".", 1)
+            return self.filter_active().filter(id=shortuuid.decode(encoded_id), prepared_cart_hash=cart_hash)
+        return self.none()
+
+
+class OrderQuerySet(BaseCartQuerySet):
     def filter_has_payment_histories(self) -> models.QuerySet[Order]:
         return self.filter_active().filter(models.Exists(PaymentHistory.objects.filter(order=models.OuterRef("id"))))
 
@@ -67,11 +163,13 @@ def filter_in_last_six_months(self) -> models.QuerySet[Order]:
         return self.filter(created_at__gte=datetime.date.today() - datetime.timedelta(days=183))
 
 
-class Order(ScanCodeMixin, BaseAbstractModel):
+class Order(PaymentPreparationMixin, ScanCodeMixin, BaseAbstractModel):
     scancode_prefix = "order"
 
     user = models.ForeignKey(UserModel, on_delete=models.PROTECT)
     name = models.TextField()
+    prepared_cart_snapshot = models.JSONField(null=True, blank=True)
+    prepared_cart_hash = models.CharField(max_length=PAYMENT_HASH_LENGTH, null=True, blank=True)
 
     payment_histories: BaseManager[PaymentHistory]
     products: BaseManager[OrderProductRelation]
@@ -234,6 +332,37 @@ class OrderProductStatus(models.TextChoices):
     def __str__(self) -> str:  # pragma: no cover
         return f"[{self.order}] {self.product} ({self.get_status_display()})"
 
+    def save(  # type: ignore[override]
+        self,
+        *,
+        force_insert: bool = False,
+        force_update: bool = False,
+        using: str | None = None,
+        update_fields: Iterable[str] | None = None,
+        clear_parent_preparation: bool = True,
+    ) -> None:
+        super().save(force_insert=force_insert, force_update=force_update, using=using, update_fields=update_fields)
+        if clear_parent_preparation:
+            self._clear_parent_payment_preparation()
+
+    def delete(self, using: str | None = None) -> None:
+        self._clear_parent_payment_preparation()
+        super().delete(using=using)
+
+    def _clear_parent_payment_preparation(self) -> None:
+        if self.status != OrderProductRelation.OrderProductStatus.pending:
+            return
+
+        # `filter().update()` — snapshot 이 존재하는 row 만 매칭하는 단일 UPDATE.
+        # 대부분의 cart 편집은 snapshot 없는 상태라 fetch 없이 0 row UPDATE 로 끝남.
+        has_snapshot = models.Q(prepared_cart_snapshot__isnull=False) | models.Q(prepared_cart_hash__isnull=False)
+        cleared = {"prepared_cart_snapshot": None, "prepared_cart_hash": None}
+
+        if self.order_id:
+            Order.objects.filter(id=self.order_id).filter(has_snapshot).update(**cleared)
+            return
+        SingleProductCart.objects.filter(order_product_relation_id=self.id).filter(has_snapshot).update(**cleared)
+
     @property
     def not_refundable_reason(self) -> str | None:
         """
@@ -280,14 +409,43 @@ def __str__(self) -> str:  # pragma: no cover
         name = self.product_option.name if self.product_option else self.custom_response
         return f"{self.product_option_group.name} - {name}"
 
+    def save(  # type: ignore[override]
+        self,
+        *,
+        force_insert: bool = False,
+        force_update: bool = False,
+        using: str | None = None,
+        update_fields: Iterable[str] | None = None,
+    ) -> None:
+        super().save(force_insert=force_insert, force_update=force_update, using=using, update_fields=update_fields)
+        self._clear_parent_payment_preparation()
+
+    def delete(self, using: str | None = None) -> None:
+        self._clear_parent_payment_preparation()
+        super().delete(using=using)
+
+    def _clear_parent_payment_preparation(self) -> None:
+        order_product_relation = self.order_product_relation
+        if order_product_relation.status != OrderProductRelation.OrderProductStatus.pending:
+            return
+        order_product_relation._clear_parent_payment_preparation()
 
-class SingleProductCart(BaseAbstractModel):
+
+class SingleProductCartQuerySet(BaseCartQuerySet):
+    pass
+
+
+class SingleProductCart(PaymentPreparationMixin, BaseAbstractModel):
     user = models.ForeignKey(UserModel, on_delete=models.PROTECT)
     order_product_relation = models.OneToOneField(
         OrderProductRelation,
         on_delete=models.PROTECT,
         related_name="single_product_cart",
     )
+    prepared_cart_snapshot = models.JSONField(null=True, blank=True)
+    prepared_cart_hash = models.CharField(max_length=PAYMENT_HASH_LENGTH, null=True, blank=True)
+
+    objects: SingleProductCartQuerySet = SingleProductCartQuerySet.as_manager()  # type: ignore[assignment, misc]
 
     history = HistoricalRecords()
 
@@ -298,9 +456,12 @@ def to_order(self) -> Order:
             name=self.order_product_relation.product.name,
             name_ko=self.order_product_relation.product.name_ko,
             name_en=self.order_product_relation.product.name_en,
+            prepared_cart_snapshot=self.prepared_cart_snapshot,
+            prepared_cart_hash=self.prepared_cart_hash,
         )
         self.order_product_relation.order = order
-        self.order_product_relation.save()
+        # cart→Order 승격은 결제 직전 단계이므로 prepared snapshot 을 유지한 채 OPR 의 parent FK 만 갱신.
+        self.order_product_relation.save(clear_parent_preparation=False)
 
         CustomerInfo.objects.filter(single_product_cart=self).update(order=order, single_product_cart=None)
 
@@ -335,9 +496,9 @@ def is_cart(self) -> typing.Literal[True]:
     def payment_histories(self) -> list[PaymentHistory]:
         return []
 
-    @functools.cached_property
-    def products(self) -> list[OrderProductRelation]:
-        return [self.order_product_relation]
+    @property
+    def products(self) -> models.QuerySet[OrderProductRelation]:
+        return OrderProductRelation.objects.filter(id=self.order_product_relation_id)
 
     @functools.cached_property
     def name(self) -> str: