austnv/remnawave-minishop
1
0
Fork 0
Code Issues 1 Pull Requests Activity

fix: heleket signature verify

Browse Source
This commit is contained in:
3252a8
2026-05-18 22:56:56 +03:00
parent e3643ee9a0
commit 707f569f62
2 changed files with 134 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/bot/payment_providers/heleket.py
+78 -12
View File
@@ -3,6 +3,7 @@ import hashlib
import hmac
import json
import logging
from collections import OrderedDict
from typing import Any, Dict, List, Optional, Tuple
from aiogram import Bot, F, Router, types
@@ -133,7 +134,12 @@ class HeleketPresentation(ProviderEnvConfig):
TELEGRAM_EMOJI: Optional[str] = None
def _serialize_for_signature(payload: Dict[str, Any]) -> str:
def _serialize_for_signature(
payload: Dict[str, Any],
*,
ensure_ascii: bool = False,
escape_slashes: bool = True,
) -> str:
"""Serialize JSON exactly the way Heleket signs it.
Heleket's PHP example uses ``json_encode`` with ``JSON_UNESCAPED_UNICODE``
@@ -141,14 +147,59 @@ def _serialize_for_signature(payload: Dict[str, Any]) -> str:
keeps unicode untouched, then we manually escape ``/`` so that the base64
payload matches the one signed on the Heleket side.
"""
encoded = json.dumps(payload, ensure_ascii=False, separators=(",", ":"))
return encoded.replace("/", "\\/")
encoded = json.dumps(payload, ensure_ascii=ensure_ascii, separators=(",", ":"))
return encoded.replace("/", "\\/") if escape_slashes else encoded
def _compute_signature(payload: Dict[str, Any], api_key: str) -> str:
body = _serialize_for_signature(payload).encode("utf-8")
def _compute_signature(
payload: Dict[str, Any],
api_key: str,
*,
ensure_ascii: bool = False,
escape_slashes: bool = True,
) -> str:
body = _serialize_for_signature(
payload,
ensure_ascii=ensure_ascii,
escape_slashes=escape_slashes,
).encode("utf-8")
b64 = base64.b64encode(body).decode("ascii")
return hashlib.md5((b64 + api_key).encode("utf-8")).hexdigest()
return hashlib.md5((b64 + str(api_key or "")).encode("utf-8")).hexdigest()
def _signature_candidates(payload: Dict[str, Any], api_key: str) -> List[Tuple[str, str, str]]:
variants = (
("php_unicode_slash", False, True),
("unicode_no_slash", False, False),
("ascii_slash", True, True),
("ascii_no_slash", True, False),
)
candidates: List[Tuple[str, str, str]] = []
seen: set[str] = set()
for name, ensure_ascii, escape_slashes in variants:
signature = _compute_signature(
payload,
api_key,
ensure_ascii=ensure_ascii,
escape_slashes=escape_slashes,
)
if signature in seen:
continue
seen.add(signature)
canonical = _serialize_for_signature(
payload,
ensure_ascii=ensure_ascii,
escape_slashes=escape_slashes,
)
candidates.append((name, signature, canonical))
return candidates
def _signature_preview(signature: str) -> str:
signature = str(signature or "")
if len(signature) <= 12:
return signature
return f"{signature[:6]}...{signature[-6:]}"
class HeleketService(HttpClientMixin):
@@ -193,11 +244,11 @@ class HeleketService(HttpClientMixin):
@property
def merchant_id(self) -> str:
return self.config.MERCHANT_ID or ""
return (self.config.MERCHANT_ID or "").strip()
@property
def api_key(self) -> str:
return self.config.API_KEY or ""
return (self.config.API_KEY or "").strip()
@property
def currency(self) -> str:
@@ -297,9 +348,25 @@ class HeleketService(HttpClientMixin):
received = payload.get("sign")
if not isinstance(received, str) or not received:
return False
data = {k: v for k, v in payload.items() if k != "sign"}
expected = _compute_signature(data, self.api_key)
return hmac.compare_digest(expected, received)
data = OrderedDict((k, v) for k, v in payload.items() if k != "sign")
candidates = _signature_candidates(data, self.api_key)
for name, expected, _canonical in candidates:
if hmac.compare_digest(expected, received):
if name != "php_unicode_slash":
logging.info("Heleket webhook: signature matched variant %s.", name)
return True
logging.warning(
"Heleket webhook: invalid signature "
"(received=%s expected=%s canonical_json_sha256=%s api_key_len=%s).",
_signature_preview(received),
[_signature_preview(expected) for _name, expected, _canonical in candidates],
[
hashlib.sha256(canonical.encode("utf-8")).hexdigest()
for _name, _expected, canonical in candidates
],
len(self.api_key),
)
return False
async def webhook_route(self, request: web.Request) -> web.Response:
if not self.configured:
@@ -330,7 +397,6 @@ class HeleketService(HttpClientMixin):
return web.Response(status=400, text="bad_request")
if self.verify_webhook_signature and not self._verify_signature(payload):
logging.warning("Heleket webhook: invalid signature.")
return web.Response(status=403, text="invalid_signature")
uuid_value = str(payload.get("uuid") or "").strip()
tests/test_security.py
+56
View File
@@ -17,6 +17,7 @@ from bot.app.web.webapp_auth import (
)
from bot.payment_providers.cryptopay import CryptoPayService
from bot.payment_providers.freekassa import FreeKassaService
from bot.payment_providers.heleket import HeleketConfig, HeleketService, _compute_signature
from bot.payment_providers.yookassa import yookassa_webhook_route
from bot.utils.request_security import request_client_ip
from config.settings import Settings
@@ -140,6 +141,61 @@ class CryptoPayServiceTests(unittest.TestCase):
self.assertFalse(service._validate_webhook_signature(b"payload", "not-a-signature"))
class HeleketServiceTests(unittest.TestCase):
def _make_service(self, api_key: str = " payment-api-key ") -> HeleketService:
service = HeleketService.__new__(HeleketService)
service.config = HeleketConfig(ENABLED=True, MERCHANT_ID="merchant", API_KEY=api_key)
return service
def test_verify_signature_accepts_php_style_webhook_with_unicode_and_nested_payload(self):
payload = {
"type": "payment",
"uuid": "1467f384-b053-42db-9066-d8a445cc52d3",
"order_id": "435",
"amount": "190.00000000",
"payment_amount": "2.61000000",
"payment_amount_usd": "2.61",
"merchant_amount": "2.55780000",
"commission": "0.05220000",
"is_final": True,
"status": "paid",
"from": "0x7876fa0152a8eceb847154297b4ffdc85e3c4bd1",
"wallet_address_uuid": None,
"network": "polygon",
"currency": "RUB",
"payer_currency": "USDT",
"payer_amount": "2.61000000",
"payer_amount_exchange_rate": "72.72073217",
"additional_data": "Подписка на 1 месяц",
"transfer_id": None,
"convert": {
"to_currency": "USDC",
"commission": "0.00000000",
"rate": "0.99870036",
"amount": "2.55447580",
},
"txid": "0x91f2a28213bf79fba51675f92a741c820ee321babe6ec48a3ae9301d31977f83",
}
payload["sign"] = _compute_signature(payload, "payment-api-key")
self.assertTrue(self._make_service()._verify_signature(payload))
def test_verify_signature_accepts_escaped_unicode_webhook_variant(self):
payload = {
"order_id": "435",
"status": "paid",
"additional_data": "Подписка на 1 месяц",
}
payload["sign"] = _compute_signature(payload, "payment-api-key", ensure_ascii=True)
self.assertTrue(self._make_service()._verify_signature(payload))
def test_verify_signature_rejects_invalid_signature(self):
service = self._make_service("payment-api-key")
self.assertFalse(service._verify_signature({"order_id": "435", "sign": "bad"}))
class WebAppSecurityTests(unittest.IsolatedAsyncioTestCase):
def test_auth_response_sets_cookies_and_does_not_return_session_token(self):
settings = SimpleNamespace(