import base64 import json import requests from django.conf import settings from django.urls import reverse from tpv.exceptions import RedsysPaymentException from tpv.models import PaymentTransaction from tpv.settings import TransactionTypes from tpv.utils import compute_signature, decode_b64_dict 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