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

"""LiqPay Checkout (CNB) API integration.

Docs: https://www.liqpay.ua/en/doc/checkout (LiqPay also mirrors most of
this at https://www.liqpay.ua/documentation/ in Ukrainian/Russian).

Structured to mirror `monobank.py` as closely as LiqPay's actual API
allows: same public function names/signatures, same
NotConfigured/Error exception pair (subclassing the generic ones in
`base.py`), same `claim_order_paid` reuse. See `base.py` for the shared
provider contract this module implements, and `registry.py` for how the
app picks between this and monobank.py.

Requires a public/private key pair issued to the merchant's LiqPay
account, set in Адмінка → Налаштування → Оплата (falls back to
LIQPAY_PUBLIC_KEY / LIQPAY_PRIVATE_KEY in .env if the admin panel setting
is empty) -- same override precedent as MONOBANK_TOKEN. Until both keys
are configured, `create_invoice` raises LiqPayNotConfigured.

*** VERIFIED vs. STRUCTURALLY-PLAUSIBLE-BUT-UNTESTED ***
No live LiqPay credentials exist in this environment, so nothing here has
been exercised against LiqPay's real servers -- only against a mocked
`requests.post`/`requests.get` (same style as this session's monobank
race-condition test). What IS solid, because it's LiqPay's well-documented
and stable core convention, independent of any one endpoint's exact
response shape:
  - the base64(JSON) `data` + base64(sha1(private_key + data + private_key))
    `signature` request-signing scheme used for every call
      (this is LiqPay's one truly stable, widely-documented mechanism)
  - the checkout redirect being a POST of `data`+`signature` to
    https://www.liqpay.ua/api/3/checkout
  - the webhook body being those same two fields as *form* fields, not
    JSON, which is why api/payments_liqpay.py parses request.form instead
    of request.get_json() like the Monobank webhook does

What is a GENUINE GUESS, flagged inline where it matters, and should be
checked against a real LiqPay sandbox before this goes live:
  - the exact field names LiqPay's status-check response uses beyond
    "status" itself (err_code / err_description names below are best-effort)
  - verify_token()'s heuristic for "these keys look valid" -- LiqPay has no
    documented no-op "ping my credentials" endpoint the way Monobank's
    /merchant/details is, so this reuses the status-check call with a
    bogus order_id and guesses at how an auth failure differs from a
    "no such order" business error in the response
"""

from __future__ import annotations

import base64
import hashlib
import hmac
import json
import logging
import secrets

import requests

from bober_bbq.config import config
from bober_bbq.models import Order, Setting
from bober_bbq.payments.base import ProviderError, ProviderNotConfigured, claim_order_paid  # noqa: F401  (re-exported)

logger = logging.getLogger(__name__)

NAME = "liqpay"

API_BASE = "https://www.liqpay.ua/api"
CHECKOUT_URL = f"{API_BASE}/3/checkout"
REQUEST_URL = f"{API_BASE}/request"


class LiqPayNotConfigured(ProviderNotConfigured):
    pass


class LiqPayError(ProviderError):
    pass


def is_sandbox() -> bool:
    # LiqPay has no separate "test token" the way Monobank does -- sandbox
    # mode is the *same* public/private key pair, with a "sandbox": 1 flag
    # added to the signed `data` payload (LiqPay then fakes the card charge
    # instead of moving real money). Mirrors monobank_test_mode's UX even
    # though the underlying mechanism is different.
    return Setting.get_bool("liqpay_sandbox_mode", False)


def _public_key() -> str:
    return Setting.get("liqpay_public_key", "") or config.LIQPAY_PUBLIC_KEY


def _private_key() -> str:
    return Setting.get("liqpay_private_key", "") or config.LIQPAY_PRIVATE_KEY


def _keys() -> tuple[str, str]:
    public_key, private_key = _public_key(), _private_key()
    if not public_key or not private_key:
        raise LiqPayNotConfigured(
            "LiqPay public/private key is not set — configure it in Адмінка → Налаштування → Оплата"
        )
    return public_key, private_key


def _sign(data_b64: str, private_key: str) -> str:
    """LiqPay's request-signing scheme: base64(sha1(private_key + data + private_key)).
    This part is stable/well-documented across every LiqPay integration guide."""
    digest = hashlib.sha1((private_key + data_b64 + private_key).encode("utf-8")).digest()
    return base64.b64encode(digest).decode("ascii")


def _encode(payload: dict) -> tuple[str, str]:
    """Returns (data_b64, signature_b64) for `payload`, adding `version`/
    `public_key`/`sandbox` the way every LiqPay call needs."""
    public_key, private_key = _keys()
    full = {"version": 3, "public_key": public_key, **payload}
    if is_sandbox():
        full["sandbox"] = 1
    data_json = json.dumps(full, separators=(",", ":"), ensure_ascii=False)
    data_b64 = base64.b64encode(data_json.encode("utf-8")).decode("ascii")
    signature_b64 = _sign(data_b64, private_key)
    return data_b64, signature_b64


def verify_webhook_signature(data_b64: str, signature_b64: str) -> bool:
    """Verify a LiqPay webhook's `signature` form field against its `data`
    form field. Unlike Monobank (separate raw-body + header, asymmetric
    ECDSA against a key Monobank publishes), LiqPay uses its own shared
    private key symmetrically: recompute the same sha1 digest and compare.
    `data_b64` here is the raw `data` form value LiqPay posted -- it plays
    the same role `raw_body` does for Monobank."""
    try:
        _, private_key = _keys()
    except LiqPayNotConfigured:
        logger.warning("Cannot verify LiqPay webhook signature: keys not set")
        return False
    if not data_b64 or not signature_b64:
        return False
    expected = _sign(data_b64, private_key)
    return hmac.compare_digest(expected, signature_b64)


def decode_webhook_data(data_b64: str) -> dict:
    """Decode the base64 `data` field LiqPay posts (JSON once decoded).
    Callers should call `verify_webhook_signature` first — this does not
    itself check authenticity."""
    return json.loads(base64.b64decode(data_b64))


def verify_token(public_key: str | None = None, private_key: str | None = None) -> tuple[bool, str]:
    """Best-effort credential check: LiqPay has no documented no-op
    "ping my keys" endpoint (unlike Monobank's /merchant/details), so this
    calls the real status-check API (see get_invoice_status) with a
    made-up order_id that can't exist, and inspects whether LiqPay
    complains about the *keys* (bad signature / unknown public_key) or
    just says "no such order" (which means the keys themselves were fine).

    Unlike Monobank's single X-Token, LiqPay needs a public+private key
    *pair* to sign anything, so this intentionally takes two optional
    arguments instead of the single `token` the generic shape in base.py
    sketches — pass neither to test whatever's currently saved, or both to
    test values before saving (mirrors the admin panel's "Перевірити"
    button, which needs to test an unsaved pair the same way
    settings_test_monobank tests an unsaved single token).

    UNVERIFIED against a real LiqPay account — the err_code strings this
    checks for ("signature", "public_key", "auth") are a best guess based
    on published LiqPay error-code naming conventions, not confirmed
    against a live response. Treat a "looks valid" result here with the
    same caution the docstring/report calls out.
    """
    public_key = public_key if public_key is not None else _public_key()
    private_key = private_key if private_key is not None else _private_key()
    if not public_key or not private_key:
        return False, "Публічний або приватний ключ не вказано"

    probe_order_id = f"verify-token-probe-{secrets.token_hex(6)}"
    payload = {"action": "status", "order_id": probe_order_id}
    try:
        full = {"version": 3, "public_key": public_key, **payload}
        data_json = json.dumps(full, separators=(",", ":"), ensure_ascii=False)
        data_b64 = base64.b64encode(data_json.encode("utf-8")).decode("ascii")
        signature_b64 = _sign(data_b64, private_key)
        resp = requests.post(REQUEST_URL, data={"data": data_b64, "signature": signature_b64}, timeout=15)
    except requests.RequestException as e:
        return False, f"Не вдалося з'єднатись з LiqPay: {e}"

    if resp.status_code != 200:
        return False, f"LiqPay повернув помилку: {resp.status_code}"

    try:
        body = resp.json()
    except ValueError:
        return False, "Неочікувана відповідь від LiqPay"

    err_code = str(body.get("err_code") or body.get("code") or "").lower()
    # Best-effort guess at which error codes mean "your keys are wrong" vs.
    # "your keys are fine, this order just doesn't exist" — unverified, see
    # docstring above.
    auth_error_markers = ("signature", "public_key", "auth", "privatekey")
    if any(marker in err_code for marker in auth_error_markers):
        return False, f"LiqPay: невірний публічний або приватний ключ ({err_code})"

    # Any other structured response (including a "payment/order not found"
    # business error) means LiqPay accepted the signature, i.e. the keys work.
    return True, "Ключі виглядають дійсними (LiqPay прийняв підпис запиту)"


def create_invoice(order: Order) -> dict:
    """Set up a payable LiqPay checkout for `order`.

    Unlike Monobank, LiqPay has no server-to-server "create invoice" API
    call at all — a LiqPay checkout is just an HTML form (data+signature)
    posted straight to https://www.liqpay.ua/api/3/checkout from the
    customer's own browser. So "creating an invoice" here is purely local
    bookkeeping: generate our own unique order_id reference (LiqPay
    requires order_id to be unique per attempt, so a retried/second
    payment for the same Order needs a fresh one — this is NOT the same as
    Order.id) and hand back a pageUrl pointing at our own tiny redirect
    route (see api/payments_liqpay.py) that renders the auto-submitting
    form. This still raises LiqPayNotConfigured/LiqPayError the same way
    Monobank's create_invoice does, so callers don't need to know which
    provider they're talking to.
    """
    _keys()  # raises LiqPayNotConfigured if either key is missing

    order_ref = f"order-{order.id}-{secrets.token_hex(4)}"
    return {
        "invoiceId": order_ref,
        "pageUrl": f"{config.BACKEND_PUBLIC_URL}/api/payments/liqpay/redirect/{order.id}",
    }


def checkout_payload_for_order(order: Order, order_ref: str) -> tuple[str, str]:
    """Builds the (data_b64, signature_b64) pair for the checkout form the
    redirect route renders. Split out from create_invoice() because the
    redirect route regenerates this fresh at click-time from the order and
    its already-persisted `liqpay_order_id`, rather than storing the signed
    payload itself."""
    cafe_name = Setting.get("cafe_name", "Bober BBQ")
    description_template = Setting.get(
        "payment_destination_template", "Оплата замовлення {number} " + cafe_name
    )
    try:
        description = description_template.format(number=order.display_number())
    except (KeyError, IndexError):
        description = f"Оплата замовлення {order.display_number()} {cafe_name}"

    payload = {
        "action": "pay",
        "amount": order.total,  # LiqPay wants a plain decimal amount in the major unit (hryvnias) — NOT cents like Monobank
        "currency": "UAH",
        "description": description,
        "order_id": order_ref,
        "result_url": f"{config.BACKEND_PUBLIC_URL}/payment/thanks?order={order.id}",
        "server_url": config.LIQPAY_WEBHOOK_URL or f"{config.BACKEND_PUBLIC_URL}/api/payments/liqpay/webhook",
    }
    return _encode(payload)


def get_invoice_status(invoice_id: str) -> dict:
    """Polls LiqPay's Status API for `invoice_id` (our order_id reference,
    as returned by create_invoice / stored in Order.liqpay_order_id).
    Returns at least {"status": str}. Raises LiqPayError on failure."""
    payload = {"action": "status", "order_id": invoice_id}
    data_b64, signature_b64 = _encode(payload)
    resp = requests.post(REQUEST_URL, data={"data": data_b64, "signature": signature_b64}, timeout=15)
    if resp.status_code != 200:
        logger.error("LiqPay get_invoice_status failed: %s %s", resp.status_code, resp.text)
        raise LiqPayError(f"LiqPay status check failed: {resp.status_code}")
    try:
        return resp.json()
    except ValueError as e:
        logger.error("LiqPay get_invoice_status returned non-JSON: %s", resp.text)
        raise LiqPayError("LiqPay status check returned an unexpected response") from e


# LiqPay's own status vocabulary (https://www.liqpay.ua/en/doc/checkout —
# "Order statuses" table). "sandbox" is the status LiqPay reports for a
# fully-completed sandbox-mode payment (mirrors "success" for real charges).
PAID_STATUSES = {"success", "sandbox"}
FAILED_STATUSES = {"failure", "error", "reversed", "expired"}