History

Livio Bieri 93c9f73a46 fix: format		2023-12-18 15:43:58 +01:00
..
migrations	fix: format	2023-12-18 15:43:58 +01:00
tests	chore: validate signature & cleanup	2023-12-18 15:43:57 +01:00
README.md	feat: shop app; billing address apis	2023-12-18 15:42:05 +01:00
__init__.py	feat: shop app; billing address apis	2023-12-18 15:42:05 +01:00
admin.py	chore: validate signature & cleanup	2023-12-18 15:43:57 +01:00
apps.py	feat: shop app; billing address apis	2023-12-18 15:42:05 +01:00
model_utils.py	fix: initial data loading	2023-12-18 15:42:41 +01:00
models.py	chore: add price help text	2023-12-18 15:43:58 +01:00
serializers.py	feat: entities API	2023-12-18 15:42:41 +01:00
services.py	chore: validate signature & cleanup	2023-12-18 15:43:57 +01:00
urls.py	chore: move shop urls	2023-12-18 15:43:57 +01:00
views.py	fix: webhook url	2023-12-18 15:43:57 +01:00

README.md

Datatrans - Proof of Concept

Setup manual steps

HMAC_KEY: https://admin.sandbox.datatrans.com/MerchSecurAdmin.jsp
BASIC_AUTH: echo -n "{merchantid}:{password}" | base64 (merchantid: 11xxxxxx): https://admin.sandbox.datatrans.com/MenuDispatch.jsp?main=1&sub=4

Code

Simple example of the payment flow with Datatrans:


from flask import Flask, request, render_template_string, jsonify, abort
import uuid
import hmac
import hashlib
import requests
import os

app = Flask(__name__)

if "HMAC_KEY" not in os.environ:
  exit("Please set the HMAC_KEY environment variable.")

if "BASIC_AUTH" not in os.environ:
  exit("Please set the BASIC_AUTH environment variable.")

# https://admin.sandbox.datatrans.com/MerchSecurAdmin.jsp
HMAC_KEY = os.environ["HMAC_KEY"]
BASIC_AUTH = os.environ["BASIC_AUTH"]
API_ENDPOINT = "https://api.sandbox.datatrans.com/v1/transactions"

LIGHTBOX_PAGE = """
<html>
<head>
<script src="https://pay.sandbox.datatrans.com/upp/payment/js/datatrans-2.0.0.js"></script>
</head>
<body>
<button id="payButton">Pay Now</button>
<script>
  var payButton = document.getElementById('payButton');
  payButton.onclick = function() {
    Datatrans.startPayment({
      transactionId: "{{ transaction_id }}",
      opened: function() { console.log('payment-form opened'); },
      loaded: function() { console.log('payment-form loaded'); },
      closed: function() { console.log('payment-page closed'); },
      error: function(errorData) { console.log('error', errorData); }
    });
  };
</script>
</body>
</html>
"""

SUCCESS_PAGE = "<html><body><h1>Payment Success</h1></body></html>"
ERROR_PAGE = "<html><body><h1>Payment Error</h1></body></html>"
CANCEL_PAGE = "<html><body><h1>Payment Cancelled</h1></body></html>"

# TODO: There is now way to test this locally, so we need to use ngrok (?)
BASE_URL = "https://89d3-2a02-21b4-9679-d800-ac5f-489-e9f6-694e.ngrok-free.app"


@app.route("/success", methods=["GET"])
def success():
  return render_template_string(SUCCESS_PAGE)


@app.route("/error", methods=["GET"])
def error():
  return render_template_string(ERROR_PAGE)


@app.route("/cancel", methods=["GET"])
def cancel():
  return render_template_string(CANCEL_PAGE)


@app.route("/init_transaction", methods=["GET"])
def init_transaction():
  # TODO
  # for debugging, it might be handy to know the
  # user who initiated the transaction
  refno = uuid.uuid4().hex

  # TODO
  # The language of user
  language = "en"

  # Transaction payload
  payload = {
    "currency": "CHF",
    "refno": refno,
    "amount": 10_00,  # 10 CHF
    "autoSettle": True,
    "language": language,
    "redirect": {
      "successUrl": f"{BASE_URL}/success",
      "errorUrl": f"{BASE_URL}/error",
      "cancelUrl": f"{BASE_URL}/cancel",
    },
    "webhook": {
      "url": f"{BASE_URL}/webhook",
    },
  }

  # Headers
  headers = {
    "Authorization": f"Basic {BASIC_AUTH}",
    "Content-Type": "application/json",
  }

  # 1. USING LIGHTBOX
  response = requests.post(API_ENDPOINT, json=payload, headers=headers)

  if response.ok:
    transaction_id = response.json().get("transactionId")
    return render_template_string(LIGHTBOX_PAGE, transaction_id=transaction_id)
  else:
    return (
      jsonify(
        {"error": "Failed to initiate transaction", "details": response.text}
      ),
      response.status_code,
    )

  # 2. USING REDIRECT
  # # Send POST request to Datatrans API
  # response = requests.post(url, json=payload, headers=headers)

  # if response.ok:
  #     transaction_id = response.json().get('transactionId')
  #     payment_url = f'https://pay.sandbox.datatrans.com/v1/start/{transaction_id}'
  #     return redirect(payment_url)
  # else:
  #     # Return error message
  #     return jsonify({"error": "Failed to initiate transaction", "details": response.text}), response.status_code


@app.route("/webhook", methods=["POST"])
def webhook():
  """
  Checks the Datatrans-Signature header of the incoming request and validates the signature:
  https://api-reference.datatrans.ch/#section/Webhook/Webhook-signing
  """
 
  # TODO Check the state here too!

  hmac_key = HMAC_KEY

  def calculate_signature(key: str, timestamp: str, payload: str) -> str:
    key_bytes = bytes.fromhex(key)
    signing_data = f"{timestamp}{payload}".encode("utf-8")
    hmac_obj = hmac.new(key_bytes, signing_data, hashlib.sha256)
    return hmac_obj.hexdigest()

  # Header format:
  # Datatrans-Signature: t={{timestamp}},s0={{signature}}
  datatrans_signature = request.headers.get("Datatrans-Signature", "")

  try:
    parts = datatrans_signature.split(",")
    timestamp = parts[0].split("=")[1]
    received_signature = parts[1].split("=")[1]

    calculated_signature = calculate_signature(
      hmac_key, timestamp, request.data.decode("utf-8")
    )

    if calculated_signature == received_signature:
      return "Signature validated.", 200
    else:
      abort(400, "Invalid signature.")
  except (IndexError, ValueError):
    abort(400, "Invalid Datatrans-Signature header.")


if __name__ == "__main__":
  app.run(debug=True, host="0.0.0.0", port=5500)

README.md

Datatrans - Proof of Concept

Setup manual steps

Links

Code