bober-bbq-bot/bober_bbq/payments/monobank.py

"""Monobank Merchant (Acquiring) API integration.

Docs: https://monobank.ua/api-docs/acquiring/

Requires a merchant acquiring token issued to the client's FOP account (see
ТЗ section 19 / 22.8), set in Адмінка → Налаштування → Оплата (falls back to
MONOBANK_TOKEN in .env if the admin panel setting is empty). Until a token is
configured, `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, Product, Setting

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"
MERCHANT_DETAILS_URL = f"{API_BASE}/api/merchant/details"

_pubkey_cache: bytes | None = None


class MonobankNotConfigured(RuntimeError):
    pass


class MonobankError(RuntimeError):
    pass


def is_test_mode() -> bool:
    return Setting.get_bool("monobank_test_mode", False)


def _token() -> str:
    # The token lives in Settings (admin-editable, no SSH needed) — .env is
    # only a fallback for a fresh deploy before anyone's touched the admin
    # panel yet. In test mode, use the separate personal test token instead
    # (Monobank has no shared/fixed test key — every client gets their own
    # via api.monobank.ua, which accepts any Luhn-valid card number and
    # never moves real money).
    if is_test_mode():
        return Setting.get("monobank_test_token", "")
    return Setting.get("monobank_token", "") or config.MONOBANK_TOKEN


def _headers() -> dict:
    token = _token()
    if not token:
        raise MonobankNotConfigured("Monobank token is not set — configure it in Адмінка → Налаштування → Оплата")
    return {"X-Token": token, "Content-Type": "application/json"}


def verify_token(token: str | None = None) -> tuple[bool, str]:
    """Pings Monobank's merchant details endpoint to check a token actually
    works, without creating a real invoice. Pass an explicit token to test
    one before saving it; omit to test whatever's currently configured."""
    check_token = token if token is not None else _token()
    if not check_token:
        return False, "Токен не вказано"
    try:
        resp = requests.get(
            MERCHANT_DETAILS_URL, headers={"X-Token": check_token}, timeout=15
        )
    except requests.RequestException as e:
        return False, f"Не вдалося з'єднатись з Monobank: {e}"

    if resp.status_code == 200:
        data = resp.json()
        name = (data.get("merchantName") or "").strip()
        return True, "Токен дійсний" + (f" ({name})" if name else "")
    if resp.status_code in (401, 403):
        return False, "Токен недійсний або відкликаний"
    return False, f"Monobank повернув помилку: {resp.status_code}"


def create_invoice(order: Order) -> dict:
    """Create a payment invoice for the given order and return
    {"invoiceId": ..., "pageUrl": ...}. Raises MonobankNotConfigured /
    MonobankError on failure."""
    cafe_name = Setting.get("cafe_name", "Bober BBQ")
    destination_template = Setting.get(
        "payment_destination_template", "Оплата замовлення {number} " + cafe_name
    )
    try:
        destination = destination_template.format(number=order.display_number())
    except (KeyError, IndexError):
        destination = f"Оплата замовлення {order.display_number()} {cafe_name}"

    merchant_paym_info = {
        "reference": order.display_number(),
        "destination": destination,
    }

    if Setting.get_bool("monobank_show_basket", True):
        product_ids = [item.product_id for item in order.items if item.product_id]
        images = {
            p.id: p.image_url
            for p in (Product.query.filter(Product.id.in_(product_ids)).all() if product_ids else [])
            if p.image_url
        }
        base_url = config.BACKEND_PUBLIC_URL.rstrip("/")
        basket = []
        for item in order.items:
            entry = {
                "name": item.product_name,
                "qty": item.quantity,
                "sum": item.product_price * 100,
                "unit": "шт.",
            }
            image_url = images.get(item.product_id)
            if image_url:
                # Monobank's payment page fetches this itself, so it must be
                # a real publicly reachable absolute URL, not our relative
                # /static/... path.
                entry["icon"] = image_url if image_url.startswith("http") else f"{base_url}{image_url}"
            basket.append(entry)
        merchant_paym_info["basketOrder"] = basket

    payload = {
        "amount": order.total * 100,
        "ccy": 980,
        "merchantPaymInfo": merchant_paym_info,
        "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:
    """Returns Monobank's webhook-signing public key as PEM bytes (the
    "key" field is itself base64 of a PEM block, not raw DER)."""
    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_pem_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"}