vbv/server/vbv_lernwelt/shop/services.py

import hashlib
import hmac

import requests
import structlog
from django.conf import settings

from vbv_lernwelt.core.admin import User
from vbv_lernwelt.shop.datatrans.datatrans_api_client import DatatransApiClient
from vbv_lernwelt.shop.models import CheckoutState

logger = structlog.get_logger(__name__)


def create_context_data_log(
    request,
    action: str,
):
    request._request.security_request_logging = "shop_view"
    request._request.type = "datatrans"
    request_trace_id = getattr(request._request, "request_trace_id", "")

    context_data = {
        "action": action,
        "session_key": (request.data.get("session_key", "") or "")[0:255].strip(),
        "ref": (request.data.get("ref", "") or "")[0:255].strip(),
        "request_trace_id": (request_trace_id or "")[0:255].strip(),
        "request_username": (
            request._request.user.username if hasattr(request._request, "user") else ""
        ),
    }

    request._request.log_additional_json_data = context_data

    log = logger.bind(label="shop", **context_data)
    return context_data, log


class InitTransactionException(Exception):
    pass


def is_signature_valid(
    signature: str,
    payload: bytes,
    hmac_key: str = settings.DATATRANS_HMAC_KEY,
):
    """
    See the docs: https://docs.datatrans.ch/docs/additional-security
    """

    try:
        timestamp = signature.split(",")[0].split("=")[1]
        s0_expected = signature.split(",")[1].split("=")[1]

        key_hex_bytes = bytes.fromhex(hmac_key)
        timestamp_bytes = bytes(timestamp, "utf-8")

        s0_actual = hmac.new(
            key_hex_bytes, timestamp_bytes + payload, hashlib.sha256
        ).hexdigest()
    except (IndexError, ValueError):
        logger.warning(
            "Invalid signature format Expected format: t=TIMESTAMP,s0=XXXX",
            signature=signature,
        )
        return False

    return s0_actual == s0_expected


def init_datatrans_transaction(
    user: User,
    refno: str,
    amount_chf_centimes: int,
    redirect_url_success: str,
    redirect_url_error: str,
    redirect_url_cancel: str,
    webhook_url: str,
    refno2: str,
    datatrans_customer_data: dict = None,
    datatrans_int_data: dict = None,
    context_data: dict = None,
    with_cembra_byjuno_invoice: bool = False,
):
    payload = {
        # We use autoSettle=True, so that we don't have to settle the transaction:
        # -> Be aware that autoSettle has implications of the possible transaction states
        "autoSettle": True,
        "amount": amount_chf_centimes,
        "currency": "CHF",
        "language": user.language,
        "refno": str(refno),
        "refno2": str(refno2),
        "webhook": {"url": webhook_url},
        "redirect": {
            "successUrl": redirect_url_success,
            "errorUrl": redirect_url_error,
            "cancelUrl": redirect_url_cancel,
        },
    }

    if with_cembra_byjuno_invoice:
        payload["paymentMethods"] = ["INT"]
    if datatrans_customer_data:
        payload["customer"] = datatrans_customer_data
    if datatrans_int_data:
        payload["INT"] = datatrans_int_data

    # add testing configuration data
    if "fakeapi" in settings.DATATRANS_API_ENDPOINT:
        payload["user_id"] = str(user.id)

    logger.info("Initiating transaction", payload=payload)

    api_client = DatatransApiClient()
    response = api_client.post_initialize_transactions(
        json_payload=payload, context_data=context_data
    )

    if response.status_code == 201:
        transaction_id = response.json()["transactionId"]
        logger.info("Transaction initiated", transaction_id=transaction_id)
        return transaction_id
    else:
        raise InitTransactionException(
            "Transaction initiation failed:",
            response.json().get("error"),
        )


def get_transaction_state(
    transaction_id: str,
) -> CheckoutState:
    response = requests.get(
        f"{settings.DATATRANS_API_ENDPOINT}/v1/transactions/{transaction_id}",
        headers={
            "Authorization": f"Basic {settings.DATATRANS_BASIC_AUTH_KEY}",
            "Content-Type": "application/json",
        },
    )

    transaction_state = response.json()["status"]

    logger.info(
        "Transaction status retrieved",
        status_code=response.status_code,
        response=transaction_state,
    )

    return datatrans_state_to_checkout_state(transaction_state)


def get_payment_url(transaction_id: str):
    return f"{settings.DATATRANS_PAY_URL}/v1/start/{transaction_id}"


def datatrans_state_to_checkout_state(datatrans_transaction_state) -> CheckoutState:
    """
    https://api-reference.datatrans.ch/#tag/v1transactions/operation/status
    """
    if datatrans_transaction_state in ["settled", "transmitted"]:
        return CheckoutState.PAID
    elif datatrans_transaction_state == "failed":
        return CheckoutState.FAILED
    elif datatrans_transaction_state == "canceled":
        return CheckoutState.CANCELED
    else:
        # An intermediate state such as "initialized", "challenge_ongoing", etc.
        # -> we don't care about those states, we only care about final states here.
        return CheckoutState.ONGOING