Some payfast Flask integration

payfast_inbound.py

from hashlib import md5

def validate_secure_form(form, provider):
    """ Does an MD5 sum against the form items """
    if provider == "payfast":
        formsignature = form["signature"]
        formdetails = [
            tuple(["m_payment_id", form["m_payment_id"]]),
            tuple(["pf_payment_id", form["pf_payment_id"]]),
            tuple(["payment_status", form["payment_status"]]),
            tuple(["item_name", form["item_name"]]),
            tuple(["item_description", form["item_description"]]),
            tuple(["amount_gross", form["amount_gross"]]),
            tuple(["amount_fee", form["amount_fee"]]),
            tuple(["amount_net", form["amount_net"]]),
            tuple(["custom_str1", form["custom_str1"]]),
            tuple(["custom_str2", form["custom_str2"]]),
            tuple(["custom_str3", form["custom_str3"]]),
            tuple(["custom_str4", form["custom_str4"]]),
            tuple(["custom_str5", form["custom_str5"]]),
            tuple(["custom_int1", form["custom_int1"]]),
            tuple(["custom_int2", form["custom_int2"]]),
            tuple(["custom_int3", form["custom_int3"]]),
            tuple(["custom_int4", form["custom_int4"]]),
            tuple(["custom_int5", form["custom_int5"]]),
            tuple(["name_first", form["name_first"]]),
            tuple(["name_last", form["name_last"]]),
            tuple(["email_address", form["email_address"]]),
            tuple(["merchant_id", form["merchant_id"]])
        ]
        # Prepare a URL encoded query with slashes
        withslash = []
        for k, vs in formdetails:
            for v in isinstance(vs, list) and vs or [vs]:
                if v is not None:
                    withslash.append(
                        (k.encode('utf-8') if isinstance(k, str) else k,
                            v.encode('utf-8') if isinstance(v, str) else v))
        withslashstr = urlencode(withslash, doseq=True)
        signature = md5(withslashstr).hexdigest()
        if formsignature == signature:
            return True
        else:
            return False

@core.route('/payments/notification/<provider>', methods=['GET', 'POST'])
def api_payments_notification(provider):
    """ Listens for an inbound transaction notification """
    if provider == "payfast" and request.method == "POST":
        if validate_secure_form(request.form, provider):
            transaction = finance.transaction_init()
            transaction["source"] = provider
            transaction["amountGross"] = request.form["amount_gross"]
            transaction["amountFee"] = request.form["amount_fee"]
            transaction["amountNet"] = request.form["amount_net"]
            transaction["amountCurrency"] = "ZAR"
            transaction["ref"] = request.form["m_payment_id"]
                # This should be our order_id handed back to us
            transaction["refExternal"] = request.form[
                "pf_payment_id"]  # Payfast internal ref
            transaction["status"] = request.form[
                "payment_status"]  # status (from lookup?)
            transaction["billingFirstName"] = request.form["name_first"]
            transaction["billingLastName"] = request.form["name_last"]
            transaction["billingEmail"] = request.form["email_address"]
            # ignore the following from the form - item_name, item_description
            if checkout.order_exists(request.form["m_payment_id"]):
                oid = request.form["m_payment_id"]
                
            else:
                transaction["notes"] = "NEEDS REVIEW: Malformed Order ID"
                
            return "SUCCESS. STORED: " + transactionkey
        else:
            raise Exception()
            g.sentry.captureMessage(
                'api_payments_notification - validate_secure_form failed')
            return "FAILED"
    g.sentry.captureMessage('api_payments_notification - provider: ' +
                            provider + ' / request.method' + request.method)
    return "FAILED"
    

payfast_outbound.py

            
@core.route('/checkout/step/pay', methods=['GET', 'POST'])
@shareddefs.session_id_required
def checkout_step_pay():
    if "order_id" in session:  # order process already started
        oid = session['order_id']  # set from session
        order = order_load(oid)  # load from db
        paydetails = [
            tuple(["merchant_id", core.config['PAYFAST_MERCHANT_ID']]),
            tuple(["merchant_key", core.config['PAYFAST_MERCHANT_KEY']]),
            tuple(["return_url", request.url_root[0:len(request.url_root)
                                                  - 1] + url_for(
                                                      'checkout_step_thanks')]),
            tuple(["cancel_url", request.url_root[0:len(request.url_root)
                                                  - 1] + url_for(
                                                      'checkout_step_cancel')]),
            tuple(["notify_url", request.url_root[0:len(request.url_root) - 1] + url_for('api_payments_notification', provider='payfast')]),
            tuple(["name_first", order["who"]["firstName"]]),
            tuple(["name_last", order["who"]["lastName"]]),
            tuple(["email_address", order["who"]["email"]]),
            tuple(["m_payment_id", session["order_id"]]),
            tuple(["amount", order["pay"]["total"]]),
            tuple(["item_name", "On The Way order: " + oid]),
            tuple(["item_description", "Your order with ontheway.co.za"])
        ]
        # Prepare a URL encoded query
        query = []
        for k, vs in paydetails:
            for v in isinstance(vs, list) and vs or [vs]:
                if v is not None:
                    query.append(
                        (k.encode('utf-8') if isinstance(k, str) else k,
                            v.encode('utf-8') if isinstance(v, str) else v))
        withslashstr = urlencode(query, doseq=True)
        signature = md5(withslashstr).hexdigest()
    else:  # go back one stage
        flash({"type": "error", "highlight": "Sorry!", "text":
              "You can't pay without going through the checkout process!"})
        return redirect(url_for('checkout_summary'))
    if "view" in session and session["view"] == "mobile":
        return render_template(core.config['TEMPLATE_ROOT']+'checkout_step_pay_mobile.html', order=order, paydetails=paydetails, signature=signature, paymenturl=core.config['PAYFAST_ENDPOINT'])
    else:
        return render_template(core.config['TEMPLATE_ROOT']+'checkout_step_pay.html', order=order, paydetails=paydetails, signature=signature, paymenturl=core.config['PAYFAST_ENDPOINT'])