fix: fix Platega and CryptoBot webhook verification

Platega: handle verification ping POST without auth headers (empty body → 200 OK)
CryptoBot: always use API token for signature verification per docs, not WEBHOOK_SECRET
CryptoBot: reject requests without signature in both FastAPI and aiohttp handlers
Remove dead self.webhook_secret from CryptoBotService
Update tests to match new behavior
This commit is contained in:
Fringg
2026-03-18 20:11:37 +03:00
parent db0e169a41
commit b40a812f3a
5 changed files with 33 additions and 20 deletions
+20 -13
View File
@@ -16,7 +16,6 @@ class CryptoBotService:
def __init__(self):
self.api_token = settings.CRYPTOBOT_API_TOKEN
self.base_url = settings.get_cryptobot_base_url()
self.webhook_secret = settings.CRYPTOBOT_WEBHOOK_SECRET
async def _make_request(
self,
@@ -123,37 +122,45 @@ class CryptoBotService:
return await self._make_request('GET', 'getExchangeRates')
def verify_webhook_signature(self, body: str, signature: str) -> bool:
# По документации CryptoBot, ключ SHA256 от API токена
token = self.webhook_secret or self.api_token
# По документации CryptoBot, ключ ВСЕГДА SHA256 от API токена
token = self.api_token
if not token:
logger.warning('CryptoBot webhook secret и API token не настроены')
logger.warning('CryptoBot API token не настроен, пропуск проверки подписи')
return True
try:
secret_hash = hashlib.sha256(token.encode()).digest()
# 1. Raw body — CryptoBot скорее всего шлёт compact JSON
# 1. Raw body — CryptoBot шлёт compact JSON
expected = hmac.new(secret_hash, body.encode('utf-8'), hashlib.sha256).hexdigest()
if hmac.compare_digest(signature, expected):
logger.info('CryptoBot webhook подпись валидна')
logger.info('CryptoBot webhook подпись валидна (raw body)')
return True
# 2. Fallback: JSON.stringify(body) как в доке — re-serialize compact
# 2. Fallback: re-serialize compact JSON
parsed = json.loads(body)
check_string = json.dumps(parsed, separators=(',', ':'), ensure_ascii=False)
expected = hmac.new(secret_hash, check_string.encode('utf-8'), hashlib.sha256).hexdigest()
if hmac.compare_digest(signature, expected):
expected_reserialized = hmac.new(secret_hash, check_string.encode('utf-8'), hashlib.sha256).hexdigest()
if hmac.compare_digest(signature, expected_reserialized):
logger.info('CryptoBot webhook подпись валидна (re-serialized)')
return True
# 3. Fallback: ensure_ascii=True (если CryptoBot эскейпит unicode)
# 3. Fallback: ensure_ascii=True
check_string_ascii = json.dumps(parsed, separators=(',', ':'), ensure_ascii=True)
expected = hmac.new(secret_hash, check_string_ascii.encode('utf-8'), hashlib.sha256).hexdigest()
if hmac.compare_digest(signature, expected):
expected_ascii = hmac.new(secret_hash, check_string_ascii.encode('utf-8'), hashlib.sha256).hexdigest()
if hmac.compare_digest(signature, expected_ascii):
logger.info('CryptoBot webhook подпись валидна (ascii-escaped)')
return True
logger.error('Неверная подпись CryptoBot webhook')
logger.error(
'Неверная подпись CryptoBot webhook',
received_signature=signature,
expected_raw=expected,
expected_reserialized=expected_reserialized,
body_length=len(body),
token_length=len(token),
token_prefix=token[:4] + '...',
)
return False
except Exception as e:
+4 -1
View File
@@ -377,7 +377,10 @@ class WebhookServer:
signature = request.headers.get('Crypto-Pay-API-Signature')
logger.info('CryptoBot Signature', signature=signature)
if signature and settings.CRYPTOBOT_WEBHOOK_SECRET:
if settings.CRYPTOBOT_API_TOKEN:
if not signature:
logger.error('CryptoBot webhook без подписи')
return web.json_response({'status': 'error', 'reason': 'missing_signature'}, status=401)
from app.external.cryptobot import CryptoBotService
cryptobot_service = CryptoBotService()
+1 -1
View File
@@ -571,7 +571,7 @@ class BotConfigurationService:
'format': 'Булево значение.',
'example': 'Включите после указания токена API и секрета вебхука.',
'warning': 'Пустой токен или неверный вебхук приведут к отказам платежей.',
'dependencies': 'CRYPTOBOT_API_TOKEN, CRYPTOBOT_WEBHOOK_SECRET',
'dependencies': 'CRYPTOBOT_API_TOKEN',
},
'PAYMENT_VERIFICATION_AUTO_CHECK_ENABLED': {
'description': (
+5 -1
View File
@@ -312,7 +312,7 @@ def create_payment_router(bot: Bot, payment_service: PaymentService) -> APIRoute
)
signature = request.headers.get('Crypto-Pay-API-Signature')
secret = settings.CRYPTOBOT_WEBHOOK_SECRET or settings.CRYPTOBOT_API_TOKEN
secret = settings.CRYPTOBOT_API_TOKEN
if secret:
if not signature:
return JSONResponse(
@@ -682,6 +682,10 @@ def create_payment_router(bot: Bot, payment_service: PaymentService) -> APIRoute
async def platega_webhook(request: Request) -> JSONResponse:
merchant_id = request.headers.get('X-MerchantId', '')
secret = request.headers.get('X-Secret', '')
raw_body = await request.body()
if not merchant_id and not secret and not raw_body.strip():
logger.info('Platega webhook verification ping (no auth headers, empty body)')
return JSONResponse({'status': 'ok'})
if merchant_id != (settings.PLATEGA_MERCHANT_ID or '') or secret != (settings.PLATEGA_SECRET or ''):
return JSONResponse(
{'status': 'error', 'reason': 'unauthorized'},
+3 -4
View File
@@ -27,7 +27,6 @@ def anyio_backend() -> str:
def _enable_token(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setattr(settings, 'CRYPTOBOT_API_TOKEN', 'token', raising=False)
monkeypatch.setattr(type(settings), 'get_cryptobot_base_url', lambda self: 'https://cryptobot.test', raising=False)
monkeypatch.setattr(settings, 'CRYPTOBOT_WEBHOOK_SECRET', 'secret', raising=False)
@pytest.mark.anyio('asyncio')
@@ -69,7 +68,7 @@ async def test_make_request_returns_none_without_token(monkeypatch: pytest.Monke
def test_verify_webhook_signature(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setattr(settings, 'CRYPTOBOT_WEBHOOK_SECRET', 'supersecret', raising=False)
monkeypatch.setattr(settings, 'CRYPTOBOT_API_TOKEN', 'supersecret', raising=False)
service = CryptoBotService()
body = '{"invoice_id":1}'
@@ -80,7 +79,7 @@ def test_verify_webhook_signature(monkeypatch: pytest.MonkeyPatch) -> None:
assert service.verify_webhook_signature(body, 'invalid') is False
def test_verify_webhook_signature_without_secret(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setattr(settings, 'CRYPTOBOT_WEBHOOK_SECRET', '', raising=False)
def test_verify_webhook_signature_without_token(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setattr(settings, 'CRYPTOBOT_API_TOKEN', '', raising=False)
service = CryptoBotService()
assert service.verify_webhook_signature('{}', 'anything') is True