import hashlib import hmac import uuid import requests import structlog from django.conf import settings from vbv_lernwelt.core.admin import User from vbv_lernwelt.shop.models import CheckoutState logger = structlog.get_logger(__name__) 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_transaction( user: User, amount_chf_centimes: int, redirect_url_success: str, redirect_url_error: str, redirect_url_cancel: str, webhook_url: str, ): if overwrite := settings.DATATRANS_DEBUG_WEBHOOK_OVERWRITE: logger.warning( "APPLYING DEBUG DATATRANS WEBHOOK OVERWRITE!", webhook_url=overwrite, ) webhook_url = overwrite 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(uuid.uuid4()), "webhook": {"url": webhook_url}, "redirect": { "successUrl": redirect_url_success, "errorUrl": redirect_url_error, "cancelUrl": redirect_url_cancel, }, } logger.info("Initiating transaction", payload=payload) response = requests.post( url=f"{settings.DATATRANS_API_ENDPOINT}/v1/transactions", json=payload, headers={ "Authorization": f"Basic {settings.DATATRANS_BASIC_AUTH_KEY}", "Content-Type": "application/json", }, ) 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