From b40a812f3aa0596bf6c5105008451dd8a17b103f Mon Sep 17 00:00:00 2001 From: Fringg Date: Wed, 18 Mar 2026 20:11:37 +0300 Subject: [PATCH] fix: fix Platega and CryptoBot webhook verification MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 --- app/external/cryptobot.py | 33 ++++++++++++++---------- app/external/webhook_server.py | 5 +++- app/services/system_settings_service.py | 2 +- app/webserver/payments.py | 6 ++++- tests/external/test_cryptobot_service.py | 7 +++-- 5 files changed, 33 insertions(+), 20 deletions(-) diff --git a/app/external/cryptobot.py b/app/external/cryptobot.py index e2f2c81e..f992fa74 100644 --- a/app/external/cryptobot.py +++ b/app/external/cryptobot.py @@ -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: diff --git a/app/external/webhook_server.py b/app/external/webhook_server.py index 346eaef8..0d0a46d9 100644 --- a/app/external/webhook_server.py +++ b/app/external/webhook_server.py @@ -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() diff --git a/app/services/system_settings_service.py b/app/services/system_settings_service.py index 53b5adb0..edc24e62 100644 --- a/app/services/system_settings_service.py +++ b/app/services/system_settings_service.py @@ -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': ( diff --git a/app/webserver/payments.py b/app/webserver/payments.py index 940bc5c5..f3ef44f9 100644 --- a/app/webserver/payments.py +++ b/app/webserver/payments.py @@ -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'}, diff --git a/tests/external/test_cryptobot_service.py b/tests/external/test_cryptobot_service.py index 96e901e5..83877291 100644 --- a/tests/external/test_cryptobot_service.py +++ b/tests/external/test_cryptobot_service.py @@ -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