fix: harden hwid provider payment edge cases

backend/bot/payment_providers/shared/__init__.py

@@ -36,10 +36,12 @@ from .common import (
     json_error,
     make_translator,
     mark_payment_failed_creation,
+    parse_positive_int_units,
     payment_failed,
     payment_link_response,
     payment_record_amounts,
     payment_unavailable,
+    payment_units_for_activation,
     sale_mode_base,
     sale_mode_is_hwid_devices,
     sale_mode_is_traffic,
@@ -103,6 +105,7 @@ __all__ = [
     "lookup_payment_by_order_or_provider_id",
     "make_translator",
     "mark_payment_failed_creation",
+    "parse_positive_int_units",
     "notify_admins_payment_received",
     "notify_callback_parse_error",
     "notify_payment_gateway_failure",
@@ -114,6 +117,7 @@ __all__ = [
     "payment_link_message_text",
     "payment_link_response",
     "payment_record_amounts",
+    "payment_units_for_activation",
     "payment_unavailable",
     "post_json_request",
     "quote_hwid_callback_parts",

backend/bot/payment_providers/shared/callbacks.py

@@ -20,6 +20,7 @@ from .common import (
     build_payment_description,
     format_human_units,
     mark_payment_failed_creation,
+    parse_positive_int_units,
     sale_mode_base,
     sale_mode_is_hwid_devices,
     sale_mode_tariff_key,
@@ -124,10 +125,13 @@ async def quote_hwid_callback_parts(
 ) -> tuple[Optional[PaymentCallbackParts], Optional[dict]]:
     if not sale_mode_is_hwid_devices(parts.sale_mode):
         return parts, None
+    device_count = parse_positive_int_units(parts.months)
+    if device_count is None:
+        return None, None
     quote = await subscription_service.quote_hwid_device_topup(
         session,
         user_id=user_id,
-        device_count=int(parts.months),
+        device_count=device_count,
         tariff_key=sale_mode_tariff_key(parts.sale_mode),
         renewal=sale_mode_base(parts.sale_mode) == "hwid_devices_renewal",
         currency=currency,
@@ -135,7 +139,7 @@ async def quote_hwid_callback_parts(
     if not quote:
         return None, None
     quoted_parts = PaymentCallbackParts(
-        months=parts.months,
+        months=device_count,
         price=float(quote.get("price") or 0),
         sale_mode=parts.sale_mode,
     )

backend/bot/payment_providers/shared/common.py

@@ -1,7 +1,7 @@
 from __future__ import annotations
 
 from dataclasses import dataclass
-from decimal import ROUND_HALF_UP, Decimal
+from decimal import ROUND_HALF_UP, Decimal, InvalidOperation
 from typing import Any, Callable, Optional
 
 from aiohttp import web
@@ -36,6 +36,20 @@ def decimal_amounts_equal(left: Any, right: Any, places: int = 2) -> bool:
     return format_decimal_amount(left, places) == format_decimal_amount(right, places)
 
 
+def parse_positive_int_units(value: Any) -> Optional[int]:
+    """Return a positive integer only when the input represents whole units exactly."""
+    if isinstance(value, bool):
+        return None
+    try:
+        decimal_value = Decimal(str(value).strip())
+    except (InvalidOperation, ValueError):
+        return None
+    if not decimal_value.is_finite() or decimal_value != decimal_value.to_integral_value():
+        return None
+    integer_value = int(decimal_value)
+    return integer_value if integer_value > 0 else None
+
+
 def format_human_units(value: Any) -> str:
     """Render numeric units the way the UI expects: integers w/o decimals, floats with %g."""
     numeric = float(value)
@@ -164,6 +178,20 @@ def payment_record_amounts(
     )
 
 
+def payment_units_for_activation(payment: Any, sale_mode: str) -> Any:
+    """Resolve purchased units from a payment record for webhook activation."""
+    base = sale_mode_base(sale_mode)
+    if sale_mode_is_traffic(base):
+        return getattr(payment, "purchased_gb", None) or getattr(
+            payment, "subscription_duration_months", None
+        ) or 1
+    if sale_mode_is_hwid_devices(base):
+        return getattr(payment, "purchased_hwid_devices", None) or getattr(
+            payment, "subscription_duration_months", None
+        ) or 1
+    return getattr(payment, "subscription_duration_months", None) or 1
+
+
 def json_error(status: int, code: str, message: str) -> web.Response:
     return web.json_response({"ok": False, "error": code, "message": message}, status=status)