fix(tradein/payments): строгий разбор нотификации и отказ вместо догадок на враждебном входе #2737
7 changed files with 545 additions and 29 deletions
|
|
@ -7,7 +7,10 @@
|
|||
делает следующий PR-D. См. `mera-tbank-acquiring-recon.md` (корень репо)
|
||||
§3/§9 для полной схемы разбивки.
|
||||
|
||||
- `token.py` — подпись `Token` запросов + проверка подписи нотификаций.
|
||||
- `token.py` — подпись `Token` запросов + проверка подписи нотификаций
|
||||
(никогда не кидает исключение на враждебном входе).
|
||||
- `notification.py` — строгий типизированный разбор тела нотификации ПОСЛЕ
|
||||
проверки подписи (`parse_notification`) — сырой `dict` дальше не уходит.
|
||||
- `receipt.py` — сборка `Receipt` (54-ФЗ, ФФД 1.05) для услуги.
|
||||
- `tbank_client.py` — httpx-клиент `Init/GetState/CheckOrder/Confirm/Cancel`.
|
||||
|
||||
|
|
|
|||
116
tradein-mvp/backend/app/services/payments/notification.py
Normal file
116
tradein-mvp/backend/app/services/payments/notification.py
Normal file
|
|
@ -0,0 +1,116 @@
|
|||
"""Строгий типизированный разбор нотификации Т-Банк — ПОСЛЕ проверки подписи.
|
||||
|
||||
Вызывать `parse_notification()` только когда `token.verify_notification_token(...)`
|
||||
уже вернул `True`. Разбор здесь НЕ проверяет подпись повторно — он только
|
||||
превращает уже доверенный (по подписи) `dict` в типизированный объект, чтобы
|
||||
сырой `dict` не утекал дальше в бизнес-логику (статус-машину платежа, запись
|
||||
в БД).
|
||||
|
||||
ПОЧЕМУ строгий разбор — самостоятельный слой защиты, а не формальность:
|
||||
алгоритм подписи Т-Банка конкатенирует значения полей БЕЗ разделителя между
|
||||
ними (см. `token.py`, docstring модуля, шаг 4). Из-за этого символы могут
|
||||
"перекладываться" между лексикографически соседними ключами так, что итоговая
|
||||
строка для SHA-256 не меняется, хотя значения полей — меняются. Проверено
|
||||
живым расчётом на официальном эталонном векторе: `Amount=1111, CardId="000000"`
|
||||
даёт тот же Token, что и `Amount=11, CardId="11000000"` (доп. `1` "перетекла"
|
||||
из `Amount` в начало `CardId`, потому что `Amount` < `CardId` лексикографически
|
||||
и обе стоят подряд в конкатенации). Значит подпись сама по себе НЕ гарантирует,
|
||||
что банк прислал именно ту сумму, которую записал у себя платёжный сервис —
|
||||
это СВОЙСТВО алгоритма банка, менять его нельзя (мы не управляем форматом
|
||||
Token, который реально пришлёт банк на проде).
|
||||
|
||||
КОНТРАКТ ДЛЯ PR-D (публичная ручка нотификации) — единственная реальная защита
|
||||
от описанного выше перекладывания: `amount_kopecks` из `parse_notification()`
|
||||
ОБЯЗАН быть сверен с уже сохранённым `payments.amount_kopecks` в БД (запись,
|
||||
созданная на `init_payment()`, найденная по `order_id`/`payment_id` из этой же
|
||||
нотификации) ДО того, как нотификация будет принята как валидное событие.
|
||||
Если сумма из нотификации не совпадает с суммой в БД — это либо подделанная
|
||||
нотификация (перекладывание символов дало другой `OrderId`/`Amount`-ключ и
|
||||
подпись всё равно сошлась), либо рассинхронизация, но НЕ штатный кейс — то и
|
||||
другое должно быть отказом, а не «примерно похоже, примем».
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass
|
||||
from typing import Any
|
||||
|
||||
|
||||
class NotificationParseError(ValueError):
|
||||
"""Поле нотификации не соответствует ожидаемому типу — отказ, не догадка."""
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class TBankNotification:
|
||||
"""Типизированное тело нотификации Т-Банка ПОСЛЕ успешной проверки подписи.
|
||||
|
||||
`amount_kopecks` здесь — то, что ПРИСЛАЛ банк в текущем HTTP-запросе, а
|
||||
НЕ подтверждённый источник истины сам по себе. См. docstring модуля —
|
||||
сверка с `payments.amount_kopecks` в БД обязательна на вызывающей стороне.
|
||||
"""
|
||||
|
||||
success: bool
|
||||
status: str
|
||||
order_id: str
|
||||
payment_id: str
|
||||
terminal_key: str
|
||||
amount_kopecks: int
|
||||
|
||||
|
||||
def parse_notification(payload: dict[str, Any]) -> TBankNotification:
|
||||
"""Строгий разбор `payload` в `TBankNotification`.
|
||||
|
||||
Вызывать ТОЛЬКО после `token.verify_notification_token(payload, password)
|
||||
is True` — эта функция подпись не проверяет.
|
||||
|
||||
Правила (без исключений, без «примерно разберём»):
|
||||
- `Success` — только настоящий `bool` (не строка `"true"`, не `1`);
|
||||
- `Amount` — только `int`; `bool` — подкласс `int` в Python
|
||||
(`isinstance(True, int) is True`), поэтому проверяется и отсекается
|
||||
ДО проверки на `int`, иначе `Success`-подобное поле молча прошло бы
|
||||
как сумма;
|
||||
- `Status`, `OrderId`, `PaymentId`, `TerminalKey` — только непустой `str`.
|
||||
|
||||
Любое несоответствие — `NotificationParseError` с указанием поля,
|
||||
ожидаемого типа и того, что реально пришло.
|
||||
"""
|
||||
if not isinstance(payload, dict):
|
||||
raise NotificationParseError(f"payload должен быть dict, получено {type(payload).__name__}")
|
||||
|
||||
return TBankNotification(
|
||||
success=_require_strict_bool(payload, "Success"),
|
||||
status=_require_nonempty_str(payload, "Status"),
|
||||
order_id=_require_nonempty_str(payload, "OrderId"),
|
||||
payment_id=_require_nonempty_str(payload, "PaymentId"),
|
||||
terminal_key=_require_nonempty_str(payload, "TerminalKey"),
|
||||
amount_kopecks=_require_strict_int(payload, "Amount"),
|
||||
)
|
||||
|
||||
|
||||
def _require_strict_bool(payload: dict[str, Any], key: str) -> bool:
|
||||
value = payload.get(key)
|
||||
if not isinstance(value, bool):
|
||||
raise NotificationParseError(
|
||||
f"{key} должен быть bool, получено {type(value).__name__}={value!r}"
|
||||
)
|
||||
return value
|
||||
|
||||
|
||||
def _require_strict_int(payload: dict[str, Any], key: str) -> int:
|
||||
value = payload.get(key)
|
||||
# bool — подкласс int в Python: проверяем и отсекаем ДО isinstance(value, int),
|
||||
# иначе True/False молча прошли бы как Amount=1/Amount=0.
|
||||
if isinstance(value, bool) or not isinstance(value, int):
|
||||
raise NotificationParseError(
|
||||
f"{key} должен быть int (не bool/str/float), получено {type(value).__name__}={value!r}"
|
||||
)
|
||||
return value
|
||||
|
||||
|
||||
def _require_nonempty_str(payload: dict[str, Any], key: str) -> str:
|
||||
value = payload.get(key)
|
||||
if not isinstance(value, str) or not value:
|
||||
raise NotificationParseError(
|
||||
f"{key} должен быть непустой str, получено {type(value).__name__}={value!r}"
|
||||
)
|
||||
return value
|
||||
|
|
@ -19,6 +19,16 @@ Docs: https://developer.tbank.ru/eacq/api
|
|||
- Бизнес-отказ (HTTP 200, но `Success: false` в теле) — тоже НЕ
|
||||
ретраится: это содержательный ответ банка, а не сбой транспорта.
|
||||
|
||||
БЮДЖЕТ ВРЕМЕНИ (важно для PR-D): worst case одного вызова любого метода —
|
||||
около 74 с (4 попытки × `_DEFAULT_TIMEOUT_S`=15 с = 60 с, плюс backoff между
|
||||
попытками 2+4+8=14 с при `_DEFAULT_MAX_RETRIES`=3). Т-Банк даёт на ответ на
|
||||
нотификацию окно порядка 10 с — этот бюджет в 74 с в него заведомо не
|
||||
укладывается. Значит: исходящий HTTP-вызов к `TBankClient` (в т.ч.
|
||||
`get_state`/`confirm`/`cancel` для сверки/реконсиляции по нотификации)
|
||||
ВНУТРИ обработчика публичной ручки нотификации ЗАПРЕЩЁН — обработчик обязан
|
||||
только валидировать/сохранить событие и ответить `"OK"`, а любая сверка с
|
||||
банком (`GetState`/`CheckOrder`) — асинхронно, вне HTTP-цикла ответа банку.
|
||||
|
||||
БЕЗОПАСНОСТЬ: `password` и `Token` НИКОГДА не попадают в `logger.*` —
|
||||
логируем только имя метода, HTTP-статус, `ErrorCode`/`Message`/`Details`
|
||||
из ответа банка.
|
||||
|
|
@ -192,7 +202,17 @@ class TBankClient:
|
|||
pay_type: str | None = None,
|
||||
data: dict[str, str] | None = None,
|
||||
) -> dict[str, Any]:
|
||||
"""`POST /v2/Init` — инициирует платёж, возвращает `PaymentId` + `PaymentURL`."""
|
||||
"""`POST /v2/Init` — инициирует платёж, возвращает `PaymentId` + `PaymentURL`.
|
||||
|
||||
КОНТРАКТ ДЛЯ PR-D (обработка сетевой ошибки вызывающей стороной):
|
||||
после `TBankApiError` от `Init` (в т.ч. `error_code == "network_error"` —
|
||||
таймаут/обрыв) НЕЛЬЗЯ слепо повторять `init_payment()` с тем же
|
||||
`order_id` — неизвестно, дошёл ли исходный запрос до банка до обрыва
|
||||
соединения. Слепой повтор может создать ВТОРОЙ холд на тот же
|
||||
`OrderId`. Разбираться нужно через `check_order(order_id=...)` —
|
||||
он возвращает уже существующие платежи по заказу — и только по его
|
||||
результату решать, нужен ли новый `Init`.
|
||||
"""
|
||||
payload: dict[str, Any] = {"OrderId": order_id, "Amount": amount_kopecks}
|
||||
if description:
|
||||
payload["Description"] = description
|
||||
|
|
@ -225,7 +245,17 @@ class TBankClient:
|
|||
amount_kopecks: int | None = None,
|
||||
receipt: dict[str, Any] | None = None,
|
||||
) -> dict[str, Any]:
|
||||
"""`POST /v2/Confirm` — подтверждение холда (двухстадийная оплата, `PayType=T`)."""
|
||||
"""`POST /v2/Confirm` — подтверждение холда (двухстадийная оплата, `PayType=T`).
|
||||
|
||||
КОНТРАКТ ДЛЯ PR-D (обработка ошибки вызывающей стороной): после
|
||||
`TBankApiError` от `Confirm` (в т.ч. сетевой таймаут) слепой вызов
|
||||
`cancel()` для того же `payment_id` ЗАПРЕЩЁН. Таймаут/обрыв мог
|
||||
прийти УЖЕ ПОСЛЕ того, как банк фактически подтвердил холд —
|
||||
`Confirm` состоялся на стороне банка, а ответ до клиента не дошёл.
|
||||
В этом случае `cancel()` вернёт клиенту уже захваченные деньги.
|
||||
Правильная последовательность: сначала `get_state(payment_id=...)`,
|
||||
и только по актуальному статусу решать, нужен ли `cancel()`.
|
||||
"""
|
||||
payload: dict[str, Any] = {"PaymentId": payment_id}
|
||||
if amount_kopecks is not None:
|
||||
payload["Amount"] = amount_kopecks
|
||||
|
|
|
|||
|
|
@ -14,8 +14,9 @@ Docs (проверено живым запросом к doc-порталу, 2026
|
|||
здесь обобщено до правила по ТИПУ значения, а не по имени ключа: любые
|
||||
вложенные объекты/массивы, будь то `Receipt`, `DATA`, `Data`, `Items`
|
||||
или `Shops`, отсекаются одинаково, потому что все они не примитивы).
|
||||
2. `bool` → `"true"`/`"false"` (нижний регистр); `int`/`float` → строка без
|
||||
экспоненциальной записи; `str` — как есть.
|
||||
2. `bool` → `"true"`/`"false"` (нижний регистр); `int` → строка через `str()`;
|
||||
`str` — как есть. `float` НЕ поддерживается — падаем явной ошибкой (формат
|
||||
дробных чисел не задокументирован Т-Банком, см. `_stringify_value`).
|
||||
3. Добавляем пару `Password: <пароль_терминала>`.
|
||||
4. Сортируем пары по имени ключа (лексикографически по строке ключа),
|
||||
конкатенируем ТОЛЬКО значения (не ключи и не имена) в одну строку.
|
||||
|
|
@ -29,30 +30,45 @@ from __future__ import annotations
|
|||
|
||||
import hashlib
|
||||
import hmac
|
||||
import logging
|
||||
from typing import Any
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
_EXCLUDED_KEYS = frozenset({"Token"})
|
||||
|
||||
|
||||
class TokenSigningError(ValueError):
|
||||
"""Поле не может быть однозначно сериализовано в подписываемую строку."""
|
||||
|
||||
|
||||
def _stringify_value(value: bool | int | float | str) -> str:
|
||||
"""Приводит плоское значение к строке по правилам Т-Банка.
|
||||
|
||||
`bool` проверяем ДО `int`: в Python `bool` — подкласс `int`
|
||||
(`isinstance(True, int) is True`), поэтому порядок веток важен —
|
||||
иначе `True` попал бы в ветку int и дал `"1"` вместо `"true"`.
|
||||
|
||||
`float` НЕ поддерживается — падаем с `TokenSigningError`, а не
|
||||
угадываем формат. Документация Т-Банка не описывает сериализацию
|
||||
дробных чисел в подписи; прежняя реализация (`format(value, "f")` +
|
||||
rstrip нулей) была неподтверждённой догадкой, и она расходится с тем,
|
||||
что реально уходит в JSON-теле запроса: `0.1 + 0.2` подписывался бы
|
||||
как `"0.3"`, а `json.dumps(0.1 + 0.2)` даёт `"0.30000000000000004"` —
|
||||
Token не соответствовал бы фактическому телу. Денежные суммы (`Amount`)
|
||||
в этом API всегда целые копейки (`int`); для денежного пути правильнее
|
||||
явно упасть на нецелом значении, чем подписать не то, что уйдёт в сеть.
|
||||
"""
|
||||
if isinstance(value, bool):
|
||||
return "true" if value else "false"
|
||||
if isinstance(value, int):
|
||||
return str(value)
|
||||
if isinstance(value, float):
|
||||
# `format(..., "f")` — фиксированная нотация, Python никогда не
|
||||
# добавляет экспоненту при presentation type 'f' (в отличие от
|
||||
# str()/repr(), которые для очень больших/малых float дают "1e+21").
|
||||
text = format(value, "f")
|
||||
if "." in text:
|
||||
text = text.rstrip("0").rstrip(".")
|
||||
return text
|
||||
raise TokenSigningError(
|
||||
f"float в подписываемых полях не поддерживается (получено {value!r}) — "
|
||||
"формат дробных чисел не описан в документации Т-Банка, см. docstring "
|
||||
"_stringify_value"
|
||||
)
|
||||
return str(value)
|
||||
|
||||
|
||||
|
|
@ -90,9 +106,36 @@ def verify_notification_token(payload: dict[str, Any], password: str) -> bool:
|
|||
Возвращает `False`, если в payload нет строкового непустого `Token`
|
||||
(нечего сравнивать) — вызывающая сторона обязана трактовать это как
|
||||
отказ в обработке нотификации, а не как «пропустить проверку».
|
||||
|
||||
НИКОГДА не поднимает исключение — на любом враждебном/мусорном входе
|
||||
(не `dict`, не-ASCII `Token`, поля, которые ломают сериализацию внутри
|
||||
`sign()`) возвращает `False`. Это обязательное свойство для публичной
|
||||
ручки нотификации (PR-D): необработанное исключение здесь — это
|
||||
неаутентифицированный HTTP 500 в ответ банку, а любой ответ, отличный
|
||||
от `"OK"`, банк трактует как временный сбой и ретраит уведомление
|
||||
почасово в течение суток. Конкретные причины двух проверок ниже:
|
||||
- `payload` не `dict` (например список) → `.get()` кинул бы
|
||||
`AttributeError` без явной проверки типа;
|
||||
- `Token` с не-ASCII символами → `hmac.compare_digest` на строках
|
||||
требует ASCII и иначе кидает `TypeError` (документированное
|
||||
ограничение stdlib, не баг).
|
||||
"""
|
||||
received_token = payload.get("Token")
|
||||
if not isinstance(received_token, str) or not received_token:
|
||||
if not isinstance(payload, dict):
|
||||
return False
|
||||
received_token = payload.get("Token")
|
||||
if not isinstance(received_token, str) or not received_token or not received_token.isascii():
|
||||
return False
|
||||
try:
|
||||
expected_token = sign(payload, password)
|
||||
except Exception:
|
||||
# Мусорное поле где-то ещё в payload (например float — см.
|
||||
# `_stringify_value`) не должно валить проверку подписи в исключение.
|
||||
logger.warning(
|
||||
"verify_notification_token: sign() упал на входящем payload — трактуем как отказ",
|
||||
exc_info=True,
|
||||
)
|
||||
return False
|
||||
try:
|
||||
return hmac.compare_digest(expected_token, received_token)
|
||||
except TypeError:
|
||||
return False
|
||||
expected_token = sign(payload, password)
|
||||
return hmac.compare_digest(expected_token, received_token)
|
||||
|
|
|
|||
|
|
@ -260,6 +260,136 @@ async def test_business_failure_success_false_raises_without_retry() -> None:
|
|||
assert calls["n"] == 1 # НЕ ретраится
|
||||
|
||||
|
||||
async def test_confirm_retries_on_5xx_then_succeeds() -> None:
|
||||
"""Денежный вызов `Confirm` ретраится на 5xx так же, как `Init`/`GetState`."""
|
||||
calls = {"n": 0}
|
||||
|
||||
def handler(request: httpx.Request) -> httpx.Response:
|
||||
calls["n"] += 1
|
||||
if calls["n"] < 3:
|
||||
return httpx.Response(502, json={"ErrorCode": "502", "Message": "bad gw"})
|
||||
return httpx.Response(200, json={"Success": True, "Status": "CONFIRMED"})
|
||||
|
||||
_install_transport(handler)
|
||||
client = _client()
|
||||
result = await client.confirm(payment_id="1")
|
||||
|
||||
assert result["Status"] == "CONFIRMED"
|
||||
assert calls["n"] == 3
|
||||
|
||||
|
||||
async def test_confirm_retries_on_network_error_then_succeeds() -> None:
|
||||
calls = {"n": 0}
|
||||
|
||||
def handler(request: httpx.Request) -> httpx.Response:
|
||||
calls["n"] += 1
|
||||
if calls["n"] < 2:
|
||||
raise httpx.ConnectError("connection refused", request=request)
|
||||
return httpx.Response(200, json={"Success": True, "Status": "CONFIRMED"})
|
||||
|
||||
_install_transport(handler)
|
||||
client = _client()
|
||||
result = await client.confirm(payment_id="1")
|
||||
|
||||
assert result["Status"] == "CONFIRMED"
|
||||
assert calls["n"] == 2
|
||||
|
||||
|
||||
async def test_confirm_gives_up_after_max_retries_on_persistent_5xx() -> None:
|
||||
def handler(request: httpx.Request) -> httpx.Response:
|
||||
return httpx.Response(500, json={"ErrorCode": "500", "Message": "boom"})
|
||||
|
||||
_install_transport(handler)
|
||||
client = _client()
|
||||
|
||||
with pytest.raises(TBankApiError) as exc_info:
|
||||
await client.confirm(payment_id="1")
|
||||
|
||||
assert exc_info.value.error_code == "500"
|
||||
|
||||
|
||||
async def test_confirm_does_not_retry_on_4xx() -> None:
|
||||
calls = {"n": 0}
|
||||
|
||||
def handler(request: httpx.Request) -> httpx.Response:
|
||||
calls["n"] += 1
|
||||
return httpx.Response(401, json={"ErrorCode": "401", "Message": "Terminal not found"})
|
||||
|
||||
_install_transport(handler)
|
||||
client = _client()
|
||||
|
||||
with pytest.raises(TBankApiError) as exc_info:
|
||||
await client.confirm(payment_id="1")
|
||||
|
||||
assert exc_info.value.error_code == "401"
|
||||
assert calls["n"] == 1 # НЕ ретраится
|
||||
|
||||
|
||||
async def test_cancel_retries_on_5xx_then_succeeds() -> None:
|
||||
"""Денежный вызов `Cancel` ретраится на 5xx так же, как `Init`/`GetState`."""
|
||||
calls = {"n": 0}
|
||||
|
||||
def handler(request: httpx.Request) -> httpx.Response:
|
||||
calls["n"] += 1
|
||||
if calls["n"] < 3:
|
||||
return httpx.Response(503, json={"ErrorCode": "503", "Message": "unavailable"})
|
||||
return httpx.Response(200, json={"Success": True, "Status": "REFUNDED"})
|
||||
|
||||
_install_transport(handler)
|
||||
client = _client()
|
||||
result = await client.cancel(payment_id="1")
|
||||
|
||||
assert result["Status"] == "REFUNDED"
|
||||
assert calls["n"] == 3
|
||||
|
||||
|
||||
async def test_cancel_retries_on_network_error_then_succeeds() -> None:
|
||||
calls = {"n": 0}
|
||||
|
||||
def handler(request: httpx.Request) -> httpx.Response:
|
||||
calls["n"] += 1
|
||||
if calls["n"] < 2:
|
||||
raise httpx.ConnectTimeout("timed out", request=request)
|
||||
return httpx.Response(200, json={"Success": True, "Status": "REFUNDED"})
|
||||
|
||||
_install_transport(handler)
|
||||
client = _client()
|
||||
result = await client.cancel(payment_id="1")
|
||||
|
||||
assert result["Status"] == "REFUNDED"
|
||||
assert calls["n"] == 2
|
||||
|
||||
|
||||
async def test_cancel_gives_up_after_max_retries_on_persistent_network_error() -> None:
|
||||
def handler(request: httpx.Request) -> httpx.Response:
|
||||
raise httpx.ConnectError("connection refused", request=request)
|
||||
|
||||
_install_transport(handler)
|
||||
client = _client()
|
||||
|
||||
with pytest.raises(TBankApiError) as exc_info:
|
||||
await client.cancel(payment_id="1")
|
||||
|
||||
assert exc_info.value.error_code == "network_error"
|
||||
|
||||
|
||||
async def test_cancel_does_not_retry_on_4xx() -> None:
|
||||
calls = {"n": 0}
|
||||
|
||||
def handler(request: httpx.Request) -> httpx.Response:
|
||||
calls["n"] += 1
|
||||
return httpx.Response(401, json={"ErrorCode": "401", "Message": "Terminal not found"})
|
||||
|
||||
_install_transport(handler)
|
||||
client = _client()
|
||||
|
||||
with pytest.raises(TBankApiError) as exc_info:
|
||||
await client.cancel(payment_id="1")
|
||||
|
||||
assert exc_info.value.error_code == "401"
|
||||
assert calls["n"] == 1 # НЕ ретраится
|
||||
|
||||
|
||||
async def test_malformed_json_response_raises_tbank_api_error() -> None:
|
||||
def handler(request: httpx.Request) -> httpx.Response:
|
||||
return httpx.Response(200, content=b"not json at all")
|
||||
|
|
|
|||
136
tradein-mvp/backend/tests/test_payments_notification.py
Normal file
136
tradein-mvp/backend/tests/test_payments_notification.py
Normal file
|
|
@ -0,0 +1,136 @@
|
|||
"""Тесты `app.services.payments.notification` — строгий разбор нотификации.
|
||||
|
||||
`parse_notification()` вызывается ПОСЛЕ `verify_notification_token(...) is True`
|
||||
(эта функция подпись не проверяет) — тесты здесь работают с payload напрямую,
|
||||
без пересчёта подписи.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import pytest
|
||||
|
||||
from app.services.payments.notification import (
|
||||
NotificationParseError,
|
||||
TBankNotification,
|
||||
parse_notification,
|
||||
)
|
||||
|
||||
_VALID_PAYLOAD = {
|
||||
"TerminalKey": "1234567890DEMO",
|
||||
"OrderId": "order-1",
|
||||
"Success": True,
|
||||
"Status": "CONFIRMED",
|
||||
"PaymentId": "0000000",
|
||||
"Amount": 111100,
|
||||
"Token": "irrelevant-here",
|
||||
}
|
||||
|
||||
|
||||
# ── happy path ────────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def test_parse_notification_happy_path_returns_typed_object() -> None:
|
||||
result = parse_notification(_VALID_PAYLOAD)
|
||||
|
||||
assert isinstance(result, TBankNotification)
|
||||
assert result.success is True
|
||||
assert result.status == "CONFIRMED"
|
||||
assert result.order_id == "order-1"
|
||||
assert result.payment_id == "0000000"
|
||||
assert result.terminal_key == "1234567890DEMO"
|
||||
assert result.amount_kopecks == 111100
|
||||
|
||||
|
||||
def test_parse_notification_ignores_extra_fields() -> None:
|
||||
"""Лишние поля (ErrorCode, CardId, Pan, ...) в payload не мешают разбору."""
|
||||
payload = {**_VALID_PAYLOAD, "ErrorCode": "0", "CardId": "000000", "Pan": "200000******0000"}
|
||||
result = parse_notification(payload)
|
||||
assert result.order_id == "order-1"
|
||||
|
||||
|
||||
# ── payload не dict ─────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def test_parse_notification_rejects_non_dict_payload() -> None:
|
||||
with pytest.raises(NotificationParseError):
|
||||
parse_notification([_VALID_PAYLOAD]) # type: ignore[arg-type]
|
||||
|
||||
|
||||
# ── Success: только настоящий bool ──────────────────────────────────────────
|
||||
|
||||
|
||||
def test_parse_notification_rejects_success_as_string_true() -> None:
|
||||
payload = {**_VALID_PAYLOAD, "Success": "true"}
|
||||
with pytest.raises(NotificationParseError):
|
||||
parse_notification(payload)
|
||||
|
||||
|
||||
def test_parse_notification_rejects_success_as_int_one() -> None:
|
||||
payload = {**_VALID_PAYLOAD, "Success": 1}
|
||||
with pytest.raises(NotificationParseError):
|
||||
parse_notification(payload)
|
||||
|
||||
|
||||
def test_parse_notification_rejects_missing_success() -> None:
|
||||
payload = {k: v for k, v in _VALID_PAYLOAD.items() if k != "Success"}
|
||||
with pytest.raises(NotificationParseError):
|
||||
parse_notification(payload)
|
||||
|
||||
|
||||
# ── Amount: только int, bool отдельно отсекается ────────────────────────────
|
||||
|
||||
|
||||
def test_parse_notification_rejects_amount_as_bool_true() -> None:
|
||||
"""`bool` — подкласс `int` в Python, поэтому отсекается ДО общей int-проверки."""
|
||||
payload = {**_VALID_PAYLOAD, "Amount": True}
|
||||
with pytest.raises(NotificationParseError):
|
||||
parse_notification(payload)
|
||||
|
||||
|
||||
def test_parse_notification_rejects_amount_as_bool_false() -> None:
|
||||
payload = {**_VALID_PAYLOAD, "Amount": False}
|
||||
with pytest.raises(NotificationParseError):
|
||||
parse_notification(payload)
|
||||
|
||||
|
||||
def test_parse_notification_rejects_amount_as_float() -> None:
|
||||
payload = {**_VALID_PAYLOAD, "Amount": 111100.0}
|
||||
with pytest.raises(NotificationParseError):
|
||||
parse_notification(payload)
|
||||
|
||||
|
||||
def test_parse_notification_rejects_amount_as_string() -> None:
|
||||
payload = {**_VALID_PAYLOAD, "Amount": "111100"}
|
||||
with pytest.raises(NotificationParseError):
|
||||
parse_notification(payload)
|
||||
|
||||
|
||||
# ── строковые поля: только непустой str ─────────────────────────────────────
|
||||
|
||||
|
||||
@pytest.mark.parametrize("field", ["Status", "OrderId", "PaymentId", "TerminalKey"])
|
||||
def test_parse_notification_rejects_empty_string_field(field: str) -> None:
|
||||
payload = {**_VALID_PAYLOAD, field: ""}
|
||||
with pytest.raises(NotificationParseError):
|
||||
parse_notification(payload)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("field", ["Status", "OrderId", "PaymentId", "TerminalKey"])
|
||||
def test_parse_notification_rejects_non_string_field(field: str) -> None:
|
||||
payload = {**_VALID_PAYLOAD, field: 12345}
|
||||
with pytest.raises(NotificationParseError):
|
||||
parse_notification(payload)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("field", ["Status", "OrderId", "PaymentId", "TerminalKey"])
|
||||
def test_parse_notification_rejects_missing_field(field: str) -> None:
|
||||
payload = {k: v for k, v in _VALID_PAYLOAD.items() if k != field}
|
||||
with pytest.raises(NotificationParseError):
|
||||
parse_notification(payload)
|
||||
|
||||
|
||||
def test_parse_notification_result_is_frozen() -> None:
|
||||
"""`TBankNotification` — frozen dataclass, случайная мутация после разбора невозможна."""
|
||||
result = parse_notification(_VALID_PAYLOAD)
|
||||
with pytest.raises(AttributeError):
|
||||
result.amount_kopecks = 1 # type: ignore[misc]
|
||||
|
|
@ -18,7 +18,9 @@ from __future__ import annotations
|
|||
|
||||
import hashlib
|
||||
|
||||
from app.services.payments.token import sign, verify_notification_token
|
||||
import pytest
|
||||
|
||||
from app.services.payments.token import TokenSigningError, sign, verify_notification_token
|
||||
|
||||
# ── эталонный вектор №1: Init ────────────────────────────────────────────────
|
||||
# Doc-портал, шаг за шагом (см. `token.py` docstring для полного описания):
|
||||
|
|
@ -132,22 +134,30 @@ def test_int_amount_stringified_without_quotes_semantics() -> None:
|
|||
assert with_int == with_str
|
||||
|
||||
|
||||
def test_float_without_leading_zero_loss_and_no_exponent() -> None:
|
||||
"""Дробное число сериализуется без экспоненты и без хвостовых нулей.
|
||||
def test_float_field_raises_instead_of_guessing_format() -> None:
|
||||
"""`float` больше не сериализуется по угадываемому формату — явный отказ.
|
||||
|
||||
Ключи после добавления Password: "A" < "Password" (лексикографически),
|
||||
поэтому конкатенация — значение A, затем значение Password.
|
||||
Раньше `format(value, "f")` + rstrip нулей выдавал для `0.1 + 0.2` строку
|
||||
"0.3", а `json.dumps(0.1 + 0.2)` реально даёт "0.30000000000000004" —
|
||||
подписывалось не то, что уходит в JSON-теле запроса. Для денежного пути
|
||||
(Amount — всегда int, копейки) правильнее упасть, чем угадать формат.
|
||||
"""
|
||||
raw = "1234.5" + "pw"
|
||||
expected = hashlib.sha256(raw.encode("utf-8")).hexdigest()
|
||||
assert sign({"A": 1234.5}, "pw") == expected
|
||||
with pytest.raises(TokenSigningError):
|
||||
sign({"A": 1234.5}, "pw")
|
||||
|
||||
|
||||
def test_large_float_has_no_exponential_notation() -> None:
|
||||
"""Очень большое число не сваливается в экспоненциальную запись (`1e+21`)."""
|
||||
raw = "1000000000000000000000" + "pw"
|
||||
expected = hashlib.sha256(raw.encode("utf-8")).hexdigest()
|
||||
assert sign({"A": 1e21}, "pw") == expected
|
||||
def test_large_float_field_also_raises() -> None:
|
||||
"""Тот же явный отказ и для значений, которые раньше ушли бы без экспоненты."""
|
||||
with pytest.raises(TokenSigningError):
|
||||
sign({"A": 1e21}, "pw")
|
||||
|
||||
|
||||
def test_amount_plus_float_sum_would_have_diverged_from_json_raises() -> None:
|
||||
"""Закрепляет мотивацию отказа: 0.1+0.2 != json.dumps(0.1+0.2) как строка."""
|
||||
computed = 0.1 + 0.2
|
||||
assert format(computed, "f").rstrip("0").rstrip(".") == "0.3"
|
||||
with pytest.raises(TokenSigningError):
|
||||
sign({"Amount": computed}, "pw")
|
||||
|
||||
|
||||
def test_none_values_are_skipped() -> None:
|
||||
|
|
@ -196,3 +206,51 @@ def test_sort_is_by_key_name_not_insertion_order() -> None:
|
|||
raw = "".join(["2", "3", "pw", "1"]) # Alpha->2, Mid->3, Password->pw, Zeta->1
|
||||
expected = hashlib.sha256(raw.encode("utf-8")).hexdigest()
|
||||
assert forward == expected
|
||||
|
||||
|
||||
# ── verify_notification_token: НИКОГДА не кидает исключение на мусоре ─────────
|
||||
#
|
||||
# После появления публичной ручки нотификации (PR-D) необработанное
|
||||
# исключение здесь = неаутентифицированный HTTP 500 в ответ банку, а любой
|
||||
# ответ, отличный от "OK", банк трактует как временный сбой и ретраит
|
||||
# нотификацию почасово в течение суток — см. docstring
|
||||
# `verify_notification_token`.
|
||||
|
||||
|
||||
def test_verify_notification_token_rejects_non_dict_payload_list() -> None:
|
||||
"""payload — список, не dict → `.get()` кинул бы AttributeError без guard'а."""
|
||||
assert verify_notification_token([{"Token": "x"}], "pw") is False # type: ignore[arg-type]
|
||||
|
||||
|
||||
def test_verify_notification_token_rejects_non_dict_payload_string() -> None:
|
||||
assert verify_notification_token("not-a-dict", "pw") is False # type: ignore[arg-type]
|
||||
|
||||
|
||||
def test_verify_notification_token_rejects_non_dict_payload_none() -> None:
|
||||
assert verify_notification_token(None, "pw") is False # type: ignore[arg-type]
|
||||
|
||||
|
||||
def test_verify_notification_token_rejects_non_ascii_token() -> None:
|
||||
"""Не-ASCII Token → `hmac.compare_digest` кинул бы TypeError без guard'а."""
|
||||
payload = {**_NOTIFICATION_VECTOR_PAYLOAD, "Token": "кириллица-не-hex-токен"}
|
||||
assert verify_notification_token(payload, _NOTIFICATION_VECTOR_PASSWORD) is False
|
||||
|
||||
|
||||
def test_verify_notification_token_rejects_non_string_token() -> None:
|
||||
payload = {**_NOTIFICATION_VECTOR_PAYLOAD, "Token": 12345}
|
||||
assert verify_notification_token(payload, _NOTIFICATION_VECTOR_PASSWORD) is False
|
||||
|
||||
|
||||
def test_verify_notification_token_does_not_raise_on_float_field() -> None:
|
||||
"""Поле-float где-то в payload (после отказа sign() от float) → False, не исключение."""
|
||||
payload = {
|
||||
"TerminalKey": "demo",
|
||||
"OrderId": "1",
|
||||
"Amount": 11.5, # float — sign() теперь явно падает на нём
|
||||
"Token": "0" * 64,
|
||||
}
|
||||
assert verify_notification_token(payload, "pw") is False
|
||||
|
||||
|
||||
def test_verify_notification_token_rejects_empty_dict() -> None:
|
||||
assert verify_notification_token({}, "pw") is False
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue