💊 capsul.org cloud compute service - python flask web application
You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

282 lines
10 KiB

import stripe
import json
import time
import decimal
import re
from time import sleep
from flask import Blueprint
from flask import make_response
from flask import request
from flask import current_app
from flask import session
from flask import redirect
from flask import url_for
from flask import jsonify
from flask import flash
from flask import render_template
from werkzeug.exceptions import abort
from capsulflask.auth import account_required
from capsulflask.db import get_model
bp = Blueprint("payment", __name__, url_prefix="/payment")
def validate_dollars():
errors = list()
dollars = None
if "dollars" not in request.form:
errors.append("dollars is required")
dollars = None
dollars = decimal.Decimal(request.form["dollars"])
errors.append("dollars must be a number")
# TODO re enable this
# if dollars and dollars < decimal.Decimal(1):
# errors.append("dollars must be >= 1")
return [errors, dollars]
@bp.route("/btcpay", methods=("GET", "POST"))
def btcpay_payment():
errors = list()
invoice_id = None
if request.method == "POST":
result = validate_dollars()
errors = result[0]
dollars = result[1]
if len(errors) == 0:
invoice = current_app.config['BTCPAY_CLIENT'].create_invoice(dict(
itemDesc="Capsul Cloud Compute",
2 years ago
current_app.logger.info(f"created btcpay invoice: {invoice}")
# print(invoice)
invoice_id = invoice["id"]
current_app.logger.info(f"created btcpay invoice_id={invoice_id} ( {session['account']}, ${dollars} )")
get_model().create_payment_session("btcpay", invoice_id, session["account"], dollars)
return redirect(invoice["url"])
for error in errors:
return render_template("btcpay.html", invoice_id=invoice_id)
def poll_btcpay_session(invoice_id):
# so you better make sure to get the invoice directly from the horses mouth!
invoice = current_app.config['BTCPAY_CLIENT'].get_invoice(invoice_id)
if invoice['currency'] != "USD":
return [400, "invalid currency"]
dollars = invoice['price']
current_app.logger.info(f"got btcpay webhook with invoice_id={invoice_id}, status={invoice['status']} dollars={dollars}")
if invoice['status'] == "paid" or invoice['status'] == "confirmed" or invoice['status'] == "complete":
success_account = get_model().consume_payment_session("btcpay", invoice_id, dollars)
if success_account:
current_app.logger.info(f"{success_account} paid ${dollars} successfully (btcpay_invoice_id={invoice_id})")
if invoice['status'] == "complete":
get_model().btcpay_invoice_resolved(invoice_id, True)
elif invoice['status'] == "expired" or invoice['status'] == "invalid":
get_model().btcpay_invoice_resolved(invoice_id, False)
return [200, "ok"]
@bp.route("/btcpay/webhook", methods=("POST",))
def btcpay_webhook():
current_app.logger.info(f"got btcpay webhook")
# IMPORTANT! there is no signature or credential for the data sent into this webhook :facepalm:
# its just a notification, thats all.
request_data = json.loads(request.data)
invoice_id = request_data['id']
result = poll_btcpay_session(invoice_id)
abort(result[0], result[1])
@bp.route("/stripe", methods=("GET", "POST"))
def stripe_payment():
errors = list()
if request.method == "POST":
result = validate_dollars()
errors = result[0]
dollars = result[1]
if len(errors) == 0:
current_app.logger.info(f"creating stripe checkout session for {session['account']}, ${dollars}")
checkout_session = stripe.checkout.Session.create(
success_url=current_app.config['BASE_URL'] + "/payment/stripe/success?session_id={CHECKOUT_SESSION_ID}",
cancel_url=current_app.config['BASE_URL'] + "/payment/stripe",
"name": "Capsul Cloud Compute",
"images": [current_app.config['BASE_URL']+"/static/capsul-product-image.png"],
"quantity": 1,
"currency": "usd",
"amount": int(dollars*100)
stripe_checkout_session_id = checkout_session['id']
current_app.logger.info(f"stripe_checkout_session_id={stripe_checkout_session_id} ( {session['account']}, ${dollars} )")
get_model().create_payment_session("stripe", stripe_checkout_session_id, session["account"], dollars)
# We can't do this because stripe requires a bunch of server-authenticated data to be sent in the hash
# of the URL. I briefly looked into reverse-engineering their proprietary javascript in order to try to figure out
# how it works and gave up after I discovered that it would require multiple complex interactions with stripe's
# servers, and it looked like they were trying to make what I was trying to do impossible.
# I never tried running the stripe proprietary javascript in a headless brower and passing the hash from the
# headless browser to the client, but I suspect it might not work anyway because they probably have thier tracking
# cookie info in there somewhere, and if the cookie doesn't match they may refuse to display the page.
#return redirect(f"https://checkout.stripe.com/pay/{stripe_checkout_session_id}")
return redirect(f"/payment/stripe/{stripe_checkout_session_id}")
for error in errors:
return render_template("stripe.html")
def redirect_to_stripe(stripe_checkout_session_id):
if stripe_checkout_session_id and not re.match(r"^[a-zA-Z0-9_=-]+$", stripe_checkout_session_id):
stripe_checkout_session_id = '___________'
response = make_response(render_template(
if stripe_checkout_session_id is not None:
response.headers['Content-Security-Policy'] = "default-src 'self' https://js.stripe.com"
return response
def stripe_checkout_session_json(stripe_checkout_session_id):
if stripe_checkout_session_id and not re.match(r"^[a-zA-Z0-9_=-]+$", stripe_checkout_session_id):
stripe_checkout_session_id = '___________'
has_redirected_already = get_model().payment_session_redirect(session['account'], stripe_checkout_session_id)
if has_redirected_already is None:
abort(404, "Not Found")
return jsonify(dict(hasRedirectedAlready=has_redirected_already))
def validate_stripe_checkout_session(stripe_checkout_session_id):
checkout_session_completed_events = stripe.Event.list(
# Check for events created in the last half hour
'gte': int(time.time() - (30 * 60)),
for event in checkout_session_completed_events.auto_paging_iter():
checkout_session = event['data']['object']
if checkout_session and 'id' in checkout_session and checkout_session['id'] == stripe_checkout_session_id:
cents = checkout_session['display_items'][0]['amount']
dollars = decimal.Decimal(cents)/100
#consume_payment_session deletes the checkout session row and inserts a payment row
# its ok to call consume_payment_session more than once because it only takes an action if the session exists
success_email = get_model().consume_payment_session("stripe", stripe_checkout_session_id, dollars)
if success_email:
return dict(email=success_email, dollars=dollars)
return None
@bp.route("/stripe/success", methods=("GET",))
def success():
stripe_checkout_session_id = request.args.get('session_id')
if not stripe_checkout_session_id:
current_app.logger.info("/payment/stripe/success returned 400: missing required URL parameter session_id")
abort(400, "missing required URL parameter session_id")
for _ in range(0, 5):
paid = validate_stripe_checkout_session(stripe_checkout_session_id)
if paid:
current_app.logger.info(f"{paid['email']} paid ${paid['dollars']} successfully (stripe_checkout_session_id={stripe_checkout_session_id})")
return redirect(url_for("console.account_balance"))
abort(400, "this checkout session is not paid yet")
# webhook is not needed
# @bp.route("/webhook", methods=("POST",))
# def webhook():
# request_data = json.loads(request.data)
# signature = request.headers.get('stripe-signature')
# try:
# event = stripe.Webhook.construct_event(
# payload=request_data,
# sig_header=signature,
# secret=current_app.config['STRIPE_WEBHOOK_SECRET']
# )
# if event['type'] == 'checkout.session.completed':
# dollars = event['data']['object']['display_items'][0]['amount']
# stripe_checkout_session_id = event['data']['object']['id']
# #consume_payment_session deletes the checkout session row and inserts a payment row
# # its ok to call consume_payment_session more than once because it only takes an action if the session exists
# success_account = get_model().consume_payment_session("stripe", stripe_checkout_session_id, dollars)
# if success_account:
# print(f"{success_account} paid ${dollars} successfully (stripe_checkout_session_id={stripe_checkout_session_id})")
# return jsonify({'status': 'success'})
# except ValueError as e:
# print("/payment/stripe/webhook returned 400: bad request", e)
# abort(400, "bad request")
# except stripe.error.SignatureVerificationError:
# print("/payment/stripe/webhook returned 400: invalid signature")
# abort(400, "invalid signature")