|
|
||
|---|---|---|
| .. | ||
| migrations | ||
| README.md | ||
| __init__.py | ||
| admin.py | ||
| apps.py | ||
| models.py | ||
| tests.py | ||
| views.py | ||
README.md
Datatrans - Proof of Concept
Setup manual steps
HMAC_KEY: https://admin.sandbox.datatrans.com/MerchSecurAdmin.jspBASIC_AUTH:echo -n "{merchantid}:{password}" | base64(merchantid:11xxxxxx): https://admin.sandbox.datatrans.com/MenuDispatch.jsp?main=1&sub=4
Links
- https://admin.sandbox.datatrans.com
- https://api-reference.datatrans.ch/#section/Idempotency
- https://docs.datatrans.ch/docs/redirect-lightbox#section-initializing-transactions
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
"""
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)