215 lines
6.9 KiB
Python
215 lines
6.9 KiB
Python
import base64
|
|
import json
|
|
from decimal import Decimal
|
|
|
|
import requests
|
|
from django.conf import settings
|
|
from django.urls import reverse
|
|
|
|
from shop.exceptions import RedsysPaymentException
|
|
from shop.models import Order, ShopSettings
|
|
from shop.settings import TransactionTypes
|
|
from shop.utils import compute_signature, decode_b64_dict
|
|
|
|
|
|
class RedsysClient:
|
|
DEBUG_ENVIRONMENT_URL = "https://sis-t.redsys.es:25443/sis/realizarPago"
|
|
PROD_ENVIRONMENT_URL = "https://sis.redsys.es/sis/realizarPago"
|
|
|
|
REST_DEBUG_ENVIRONMENT_URL = (
|
|
"https://sis-t.redsys.es:25443/sis/rest/trataPeticionREST"
|
|
)
|
|
REST_PROD_ENVIRONMENT_URL = "https://sis.redsys.es/sis/rest/trataPeticionREST"
|
|
|
|
def __init__(self):
|
|
self.settings = ShopSettings.load()
|
|
|
|
def get_target_url(self):
|
|
shop_settings = ShopSettings.load()
|
|
if shop_settings.debug:
|
|
return self.DEBUG_ENVIRONMENT_URL
|
|
return self.PROD_ENVIRONMENT_URL
|
|
|
|
def get_rest_target_url(self):
|
|
shop_settings = ShopSettings.load()
|
|
if shop_settings.debug:
|
|
return self.REST_DEBUG_ENVIRONMENT_URL
|
|
return self.REST_PROD_ENVIRONMENT_URL
|
|
|
|
def get_merchant_code(self) -> str:
|
|
return self.settings.merchant_code
|
|
|
|
def get_terminal(self) -> str:
|
|
return self.settings.terminal
|
|
|
|
def get_currency_code(self) -> str:
|
|
return self.settings.currency_code
|
|
|
|
def get_merchant_url_ok_for_order(self, order: Order) -> str:
|
|
path = reverse("web:order", kwargs={"uuid": order.uuid})
|
|
return f"{self.settings.tpv_domain}{path}"
|
|
|
|
def to_integer(self, value: Decimal):
|
|
return int(value * 100)
|
|
|
|
def get_merchant_url_ko_for_order(self, order: Order) -> str:
|
|
path = reverse("web:order", kwargs={"uuid": order.uuid})
|
|
return f"{self.settings.tpv_domain}{path}"
|
|
|
|
def get_webhook_url_for_order(self, order: Order) -> str:
|
|
path = reverse("shop:webhook", kwargs={"uuid": order.uuid})
|
|
return f"{self.settings.tpv_domain}{path}"
|
|
|
|
def get_shared_secret(self) -> str:
|
|
return self.settings.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_order(
|
|
self,
|
|
order: Order,
|
|
order_type: int = TransactionTypes.AUTHORIZATION,
|
|
) -> dict:
|
|
merchant_code = self.get_merchant_code()
|
|
|
|
return {
|
|
"DS_MERCHANT_AMOUNT": str(self.to_integer(order.total)),
|
|
"DS_MERCHANT_CURRENCY": self.get_currency_code(),
|
|
"DS_MERCHANT_MERCHANTCODE": merchant_code,
|
|
# Webhook
|
|
"DS_MERCHANT_MERCHANTURL": self.get_webhook_url_for_order(order),
|
|
"DS_MERCHANT_ORDER": order.code,
|
|
"DS_MERCHANT_TERMINAL": self.get_terminal(),
|
|
"DS_MERCHANT_TRANSACTIONTYPE": order_type,
|
|
# Página informativa al usuario - Pago erróneo
|
|
"DS_MERCHANT_URLKO": self.get_merchant_url_ko_for_order(order),
|
|
# Página informativa al usuario - Pago confirmado
|
|
"DS_MERCHANT_URLOK": self.get_merchant_url_ok_for_order(order),
|
|
}
|
|
|
|
def _get_encoded_merchant_parameters_for_order(
|
|
self,
|
|
order: Order,
|
|
order_type: int = TransactionTypes.AUTHORIZATION,
|
|
) -> str:
|
|
body = self._get_merchant_parameters_for_order(order, order_type)
|
|
return self._encode_body(body)
|
|
|
|
def get_body_for_order(
|
|
self,
|
|
order: Order,
|
|
order_type: int = TransactionTypes.AUTHORIZATION,
|
|
) -> dict:
|
|
merchant_parameters = self._get_encoded_merchant_parameters_for_order(
|
|
order, order_type
|
|
)
|
|
signature = self.get_signature(order.code, 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_order(
|
|
self,
|
|
order: Order,
|
|
pan: str,
|
|
expiry_date: str,
|
|
cvv2: str,
|
|
order_type: int = TransactionTypes.AUTHORIZATION,
|
|
) -> dict:
|
|
merchant_code = self.get_merchant_code()
|
|
|
|
return {
|
|
"DS_MERCHANT_AMOUNT": str(self.to_integer(order.total)),
|
|
"DS_MERCHANT_CURRENCY": self.get_currency_code(),
|
|
"DS_MERCHANT_CVV2": cvv2,
|
|
"DS_MERCHANT_EXPIRYDATE": expiry_date,
|
|
"DS_MERCHANT_MERCHANTCODE": merchant_code,
|
|
"DS_MERCHANT_ORDER": order.uuid.hex,
|
|
"DS_MERCHANT_PAN": pan,
|
|
"DS_MERCHANT_TERMINAL": "1",
|
|
"DS_MERCHANT_TRANSACTIONTYPE": order_type,
|
|
}
|
|
|
|
def _get_rest_encoded_merchant_parameters_for_order(
|
|
self,
|
|
order: Order,
|
|
pan: str,
|
|
expiry_date: str,
|
|
cvv2: str,
|
|
order_type: int = TransactionTypes.AUTHORIZATION,
|
|
) -> str:
|
|
body = self._get_rest_merchant_parameters_for_order(
|
|
order,
|
|
pan=pan,
|
|
expiry_date=expiry_date,
|
|
cvv2=cvv2,
|
|
order_type=order_type,
|
|
)
|
|
return self._encode_body(body)
|
|
|
|
def _get_rest_body_for_order(
|
|
self,
|
|
order: Order,
|
|
pan: str,
|
|
expiry_date: str,
|
|
cvv2: str,
|
|
order_type: int = TransactionTypes.AUTHORIZATION,
|
|
) -> dict:
|
|
merchant_parameters = self._get_rest_encoded_merchant_parameters_for_order(
|
|
order=order,
|
|
pan=pan,
|
|
expiry_date=expiry_date,
|
|
cvv2=cvv2,
|
|
order_type=order_type,
|
|
)
|
|
signature = self.get_signature(order.code, merchant_parameters)
|
|
|
|
return {
|
|
"Ds_MerchantParameters": merchant_parameters,
|
|
"Ds_SignatureVersion": "HMAC_SHA256_V1",
|
|
"Ds_Signature": signature,
|
|
}
|
|
|
|
def make_request_for_order(
|
|
self,
|
|
order: Order,
|
|
pan: str,
|
|
expiry_date: str,
|
|
cvv2: str,
|
|
order_type: int = TransactionTypes.AUTHORIZATION,
|
|
):
|
|
body = self._get_rest_body_for_order(
|
|
order=order,
|
|
pan=pan,
|
|
expiry_date=expiry_date,
|
|
cvv2=cvv2,
|
|
order_type=order_type,
|
|
)
|
|
url = self.get_rest_target_url()
|
|
response = requests.post(url, body)
|
|
return response
|
|
|
|
def pay_order_rest(self, order: Order, pan: str, expiry_date: str, cvv2: str):
|
|
response = self.make_request_for_order(order, 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
|