byrsapty
0eabebaca3
Flask API/admin backend, aiogram bot with delivery/pickup FSM flows, monobank payment integration, and a Vite/React Telegram Mini App for menu browsing and cart management.
120 lines
4.2 KiB
Python
120 lines
4.2 KiB
Python
"""Monobank Merchant (Acquiring) API integration.
|
||
|
||
Docs: https://monobank.ua/api-docs/acquiring/
|
||
|
||
Requires MONOBANK_TOKEN in .env — the merchant acquiring token issued to the
|
||
client's FOP account (see ТЗ section 19 / 22.8). Until that token is
|
||
provided, `create_invoice` raises MonobankNotConfigured so the bot can fall
|
||
back to "pay on delivery/pickup" instead of crashing.
|
||
"""
|
||
|
||
import base64
|
||
import logging
|
||
|
||
import requests
|
||
|
||
from bober_bbq.config import config
|
||
from bober_bbq.models import Order
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
API_BASE = "https://api.monobank.ua"
|
||
CREATE_INVOICE_URL = f"{API_BASE}/api/merchant/invoice/create"
|
||
INVOICE_STATUS_URL = f"{API_BASE}/api/merchant/invoice/status"
|
||
PUBLIC_KEY_URL = f"{API_BASE}/api/merchant/pubkey"
|
||
|
||
_pubkey_cache: bytes | None = None
|
||
|
||
|
||
class MonobankNotConfigured(RuntimeError):
|
||
pass
|
||
|
||
|
||
class MonobankError(RuntimeError):
|
||
pass
|
||
|
||
|
||
def _headers() -> dict:
|
||
if not config.MONOBANK_TOKEN:
|
||
raise MonobankNotConfigured("MONOBANK_TOKEN is not set — ask the client for FOP acquiring credentials")
|
||
return {"X-Token": config.MONOBANK_TOKEN, "Content-Type": "application/json"}
|
||
|
||
|
||
def create_invoice(order: Order) -> dict:
|
||
"""Create a payment invoice for the given order and return
|
||
{"invoiceId": ..., "pageUrl": ...}. Raises MonobankNotConfigured /
|
||
MonobankError on failure."""
|
||
basket = [
|
||
{
|
||
"name": item.product_name,
|
||
"qty": item.quantity,
|
||
"sum": item.product_price * 100,
|
||
"unit": "шт.",
|
||
}
|
||
for item in order.items
|
||
]
|
||
|
||
payload = {
|
||
"amount": order.total * 100,
|
||
"ccy": 980,
|
||
"merchantPaymInfo": {
|
||
"reference": order.display_number(),
|
||
"destination": f"Оплата замовлення {order.display_number()} Bober BBQ",
|
||
"basketOrder": basket,
|
||
},
|
||
"redirectUrl": f"{config.BACKEND_PUBLIC_URL}/payment/thanks?order={order.id}",
|
||
"webHookUrl": config.MONOBANK_WEBHOOK_URL or f"{config.BACKEND_PUBLIC_URL}/api/payments/monobank/webhook",
|
||
"validity": 3600,
|
||
}
|
||
|
||
resp = requests.post(CREATE_INVOICE_URL, json=payload, headers=_headers(), timeout=15)
|
||
if resp.status_code != 200:
|
||
logger.error("Monobank create_invoice failed: %s %s", resp.status_code, resp.text)
|
||
raise MonobankError(f"Monobank invoice creation failed: {resp.status_code}")
|
||
return resp.json()
|
||
|
||
|
||
def get_invoice_status(invoice_id: str) -> dict:
|
||
resp = requests.get(INVOICE_STATUS_URL, params={"invoiceId": invoice_id}, headers=_headers(), timeout=15)
|
||
if resp.status_code != 200:
|
||
logger.error("Monobank get_invoice_status failed: %s %s", resp.status_code, resp.text)
|
||
raise MonobankError(f"Monobank status check failed: {resp.status_code}")
|
||
return resp.json()
|
||
|
||
|
||
def _fetch_public_key() -> bytes:
|
||
global _pubkey_cache
|
||
if _pubkey_cache is not None:
|
||
return _pubkey_cache
|
||
resp = requests.get(PUBLIC_KEY_URL, headers=_headers(), timeout=15)
|
||
resp.raise_for_status()
|
||
key_b64 = resp.json()["key"]
|
||
_pubkey_cache = base64.b64decode(key_b64)
|
||
return _pubkey_cache
|
||
|
||
|
||
def verify_webhook_signature(raw_body: bytes, signature_b64: str) -> bool:
|
||
"""Verify the `X-Sign` header monobank attaches to webhook requests
|
||
(ECDSA/SHA256 over the raw JSON body, public key served by monobank)."""
|
||
try:
|
||
from cryptography.exceptions import InvalidSignature
|
||
from cryptography.hazmat.primitives import hashes, serialization
|
||
from cryptography.hazmat.primitives.asymmetric import ec
|
||
|
||
public_key = serialization.load_der_public_key(_fetch_public_key())
|
||
signature = base64.b64decode(signature_b64)
|
||
public_key.verify(signature, raw_body, ec.ECDSA(hashes.SHA256()))
|
||
return True
|
||
except MonobankNotConfigured:
|
||
logger.warning("Cannot verify monobank webhook signature: MONOBANK_TOKEN not set")
|
||
return False
|
||
except InvalidSignature:
|
||
return False
|
||
except Exception:
|
||
logger.exception("Monobank webhook signature verification error")
|
||
return False
|
||
|
||
|
||
# Maps monobank invoice status -> our internal payment_status
|
||
PAID_STATUSES = {"success"}
|
||
FAILED_STATUSES = {"failure", "reversed", "expired"}
|