shoppy/tpv/redsys.py

import base64
import json
import requests

from django.conf import settings
from django.urls import reverse

from tpv.models import PaymentTransaction
from tpv.utils import compute_signature, decode_b64_dict
from tpv.settings import TransactionTypes

from tpv.exceptions import RedsysPaymentException


class RedsysClient:
    DEBUG_ENVIRONMENT_URL = "https://sis-t.redsys.es:25443/sis/realizarPago"
    PROD_ENVIRONMENT_URL = ""

    REST_DEBUG_ENVIRONMENT_URL = (
        "https://sis-t.redsys.es:25443/sis/rest/trataPeticionREST"
    )
    REST_PROD_ENVIRONMENT_URL = ""

    def get_target_url(self):
        if settings.DEBUG:
            return self.DEBUG_ENVIRONMENT_URL
        return self.PROD_ENVIRONMENT_URL

    def get_rest_target_url(self):
        if settings.DEBUG:
            return self.REST_DEBUG_ENVIRONMENT_URL
        return self.REST_PROD_ENVIRONMENT_URL

    def get_merchant_code(self) -> str:
        return settings.REDSYS_MERCHANT_CODE

    def get_terminal(self) -> str:
        return settings.REDSYS_TERMINAL

    def get_currency_code(self) -> str:
        return settings.REDSYS_CURRENCY_CODE

    def get_merchant_url_ok_for_transaction(
        self, transaction: PaymentTransaction
    ) -> str:
        path = reverse("tpv:ok", kwargs={"transaction": transaction.hash})
        return f"{settings.REDSYS_TPV_DOMAIN}{path}"

    def get_merchant_url_ko_for_transaction(
        self, transaction: PaymentTransaction
    ) -> str:
        path = reverse("tpv:ko", kwargs={"transaction": transaction.hash})
        return f"{settings.REDSYS_TPV_DOMAIN}{path}"

    def get_webhook_url_for_transaction(self, transaction: PaymentTransaction) -> str:
        path = reverse("tpv:webhook", kwargs={"transaction": transaction.hash})
        return f"{settings.REDSYS_TPV_DOMAIN}{path}"

    def get_shared_secret(self) -> str:
        return settings.REDSYS_SHARED_SECRET

    def get_signature(self, hash, payload: str):
        key = self.get_shared_secret()
        return compute_signature(str(hash), payload, key).decode()

    def _get_merchant_parameters_for_transaction(
        self,
        transaction: PaymentTransaction,
        transaction_type: int = TransactionTypes.AUTHORIZATION,
    ) -> dict:
        merchant_code = self.get_merchant_code()

        return {
            "DS_MERCHANT_AMOUNT": str(transaction.amount_integer),
            "DS_MERCHANT_CURRENCY": self.get_currency_code(),
            "DS_MERCHANT_MERCHANTCODE": merchant_code,
            # Webhook
            "DS_MERCHANT_MERCHANTURL": self.get_webhook_url_for_transaction(
                transaction
            ),
            "DS_MERCHANT_ORDER": transaction.hash.hex,
            "DS_MERCHANT_TERMINAL": self.get_terminal(),
            "DS_MERCHANT_TRANSACTIONTYPE": transaction_type,
            # Página informativa al usuario - Pago erróneo
            "DS_MERCHANT_URLKO": self.get_merchant_url_ko_for_transaction(transaction),
            # Página informativa al usuario - Pago confirmado
            "DS_MERCHANT_URLOK": self.get_merchant_url_ok_for_transaction(transaction),
        }

    def _get_encoded_merchant_parameters_for_transaction(
        self,
        transaction: PaymentTransaction,
        transaction_type: int = TransactionTypes.AUTHORIZATION,
    ) -> str:
        body = self._get_merchant_parameters_for_transaction(
            transaction, transaction_type
        )
        return self._encode_body(body)

    def get_body_for_transaction(
        self,
        transaction: PaymentTransaction,
        transaction_type: int = TransactionTypes.AUTHORIZATION,
    ) -> dict:
        merchant_parameters = self._get_encoded_merchant_parameters_for_transaction(
            transaction, transaction_type
        )
        signature = self.get_signature(transaction.hash.hex, merchant_parameters)

        return {
            "Ds_MerchantParameters": merchant_parameters,
            "Ds_SignatureVersion": "HMAC_SHA256_V1",
            "Ds_Signature": signature,
        }

    def _encode_body(self, value):
        stringified_body = json.dumps(value)
        return base64.b64encode(stringified_body.encode()).decode("utf-8")

    def _get_rest_merchant_parameters_for_transaction(
        self,
        transaction: PaymentTransaction,
        pan: str,
        expiry_date: str,
        cvv2: str,
        transaction_type: int = TransactionTypes.AUTHORIZATION,
    ) -> dict:
        merchant_code = self.get_merchant_code()

        return {
            "DS_MERCHANT_AMOUNT": str(transaction.amount_integer),
            "DS_MERCHANT_CURRENCY": self.get_currency_code(),
            "DS_MERCHANT_CVV2": cvv2,
            "DS_MERCHANT_EXPIRYDATE": expiry_date,
            "DS_MERCHANT_MERCHANTCODE": merchant_code,
            "DS_MERCHANT_ORDER": transaction.hash.hex,
            "DS_MERCHANT_PAN": pan,
            "DS_MERCHANT_TERMINAL": "1",
            "DS_MERCHANT_TRANSACTIONTYPE": transaction_type,
        }

    def _get_rest_encoded_merchant_parameters_for_transaction(
        self,
        transaction: PaymentTransaction,
        pan: str,
        expiry_date: str,
        cvv2: str,
        transaction_type: int = TransactionTypes.AUTHORIZATION,
    ) -> str:
        body = self._get_rest_merchant_parameters_for_transaction(
            transaction,
            pan=pan,
            expiry_date=expiry_date,
            cvv2=cvv2,
            transaction_type=transaction_type,
        )
        return self._encode_body(body)

    def _get_rest_body_for_transaction(
        self,
        transaction: PaymentTransaction,
        pan: str,
        expiry_date: str,
        cvv2: str,
        transaction_type: int = TransactionTypes.AUTHORIZATION,
    ) -> dict:
        merchant_parameters = (
            self._get_rest_encoded_merchant_parameters_for_transaction(
                transaction=transaction,
                pan=pan,
                expiry_date=expiry_date,
                cvv2=cvv2,
                transaction_type=transaction_type,
            )
        )
        signature = self.get_signature(transaction.hash.hex, merchant_parameters)

        return {
            "Ds_MerchantParameters": merchant_parameters,
            "Ds_SignatureVersion": "HMAC_SHA256_V1",
            "Ds_Signature": signature,
        }

    def make_request_for_transaction(
        self,
        transaction: PaymentTransaction,
        pan: str,
        expiry_date: str,
        cvv2: str,
        transaction_type: int = TransactionTypes.AUTHORIZATION,
    ):
        body = self._get_rest_body_for_transaction(
            transaction=transaction,
            pan=pan,
            expiry_date=expiry_date,
            cvv2=cvv2,
            transaction_type=transaction_type,
        )
        url = self.get_rest_target_url()
        response = requests.post(url, body)
        return response

    def pay_transaction_rest(
        self, transaction: PaymentTransaction, pan: str, expiry_date: str, cvv2: str
    ):
        response = self.make_request_for_transaction(
            transaction, pan, expiry_date, cvv2
        )
        data = response.json()
        error_code: str = data.get("errorCode", "")

        if error_code:
            error_code.replace("SIS0", "0")

            raise RedsysPaymentException(error_code)

        merchant_parameters = data.get("Ds_MerchantParameters")
        parameters = decode_b64_dict(merchant_parameters)

        return parameters