fix: PayPear webhook signature — strip signature field before hashing + IP fallback

The old code hashed the full raw body INCLUDING the 'signature' field
itself — a circular computation that can never match (you can't include
the signature in the data being signed).

Fix:
1. Strip 'signature' key from payload before HMAC-SHA256 computation
2. Try both sorted and unsorted keys (PayPear docs don't specify)
3. Fallback to IP allowlist check (158.160.85.101 per PayPear docs)
4. Pass client_ip from request headers to the verification function
This commit is contained in:
Fringg
2026-04-29 11:23:20 +03:00
parent 06db393488
commit 443a826402
2 changed files with 54 additions and 19 deletions
+47 -17
View File
@@ -208,27 +208,57 @@ class PayPearService:
logger.exception('PayPear API connection error', error=e)
raise
def verify_webhook_signature(self, raw_body: bytes, received_signature: str) -> bool:
"""Верификация подписи webhook PayPear через HMAC-SHA256.
# PayPear documented webhook source IPs
WEBHOOK_ALLOWED_IPS: set[str] = {'158.160.85.101'}
PayPear sends signature in the webhook JSON field 'signature'.
The signature is HMAC-SHA256(secret_key, raw_body).
def verify_webhook_signature(self, raw_body: bytes, received_signature: str, client_ip: str | None = None) -> bool:
"""Верификация webhook PayPear.
PayPear documentation does not specify the exact signature algorithm.
We try HMAC-SHA256(secret_key, body_without_signature_field) — the most common pattern.
If signature verification fails, fall back to IP allowlist check (recommended by PayPear docs).
"""
try:
if not received_signature:
logger.warning('PayPear webhook: отсутствует signature')
return False
import json as json_mod
expected = hmac.new(
self.secret_key.encode('utf-8'),
raw_body,
hashlib.sha256,
).hexdigest()
# Try signature verification (body without 'signature' field, sorted keys, compact separators)
if received_signature and self.secret_key:
try:
payload = json_mod.loads(raw_body)
payload_without_sig = {k: v for k, v in payload.items() if k != 'signature'}
body_to_sign = json_mod.dumps(payload_without_sig, separators=(',', ':'), sort_keys=True).encode(
'utf-8'
)
return hmac.compare_digest(expected, received_signature)
except Exception as e:
logger.error('PayPear webhook verify error', error=e)
return False
expected = hmac.new(
self.secret_key.encode('utf-8'),
body_to_sign,
hashlib.sha256,
).hexdigest()
if hmac.compare_digest(expected, received_signature):
return True
# Try without sort_keys (original key order)
body_to_sign_unsorted = json_mod.dumps(payload_without_sig, separators=(',', ':')).encode('utf-8')
expected_unsorted = hmac.new(
self.secret_key.encode('utf-8'),
body_to_sign_unsorted,
hashlib.sha256,
).hexdigest()
if hmac.compare_digest(expected_unsorted, received_signature):
return True
logger.debug('PayPear signature mismatch, falling back to IP check')
except Exception as e:
logger.debug('PayPear signature verify error, falling back to IP check', error=e)
# Fallback: IP allowlist (recommended by PayPear docs)
if client_ip and client_ip in self.WEBHOOK_ALLOWED_IPS:
return True
logger.warning('PayPear webhook: signature mismatch and IP not in allowlist', client_ip=client_ip)
return False
# Singleton instance
+7 -2
View File
@@ -1268,8 +1268,13 @@ def create_payment_router(bot: Bot, payment_service: PaymentService) -> APIRoute
from app.services.paypear_service import paypear_service
if not paypear_service.verify_webhook_signature(raw_body, received_signature):
logger.warning('PayPear webhook: invalid signature')
client_ip = (
request.headers.get('x-real-ip')
or request.headers.get('x-forwarded-for', '').split(',')[0].strip()
or (request.client.host if request.client else None)
)
if not paypear_service.verify_webhook_signature(raw_body, received_signature, client_ip=client_ip):
logger.warning('PayPear webhook: invalid signature and IP', client_ip=client_ip)
return JSONResponse({'status': False}, status_code=status.HTTP_403_FORBIDDEN)
try: