XF/envelope
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Renew with 2FA and Podman

Browse Source
This commit is contained in:
Mark Veidemanis
2025-02-07 21:53:52 +00:00
parent 906f34f3a2
commit 83a3761d17
59 changed files with 1001 additions and 861 deletions
Download Patch File Download Diff File Expand all files Collapse all files

88
core/views/base.py
View File

@@ -1,6 +1,6 @@
import logging
import stripe
# import stripe
from django.conf import settings
from django.contrib.auth.mixins import LoginRequiredMixin
from django.http import JsonResponse
@@ -10,8 +10,7 @@ from django.views import View
from django.views.generic.edit import CreateView
from core.forms import NewUserForm
from core.lib.products import assemble_plan_map
from core.models import Plan, Session
from core.lib.notify import raw_sendmsg
logger = logging.getLogger(__name__)
@@ -25,81 +24,22 @@ class Home(View):
return render(request, self.template_name)
class Billing(LoginRequiredMixin, View):
template_name = "billing.html"
def get(self, request):
if not settings.STRIPE_ENABLED:
return redirect(reverse("home"))
plans = Plan.objects.all()
user_plans = request.user.plans.all()
context = {"plans": plans, "user_plans": user_plans}
return render(request, self.template_name, context)
class Order(LoginRequiredMixin, View):
def get(self, request, plan_name):
if not settings.STRIPE_ENABLED:
return redirect(reverse("home"))
plan = Plan.objects.get(name=plan_name)
try:
cast = {
"payment_method_types": settings.ALLOWED_PAYMENT_METHODS,
"mode": "subscription",
"customer": request.user.stripe_id,
"line_items": assemble_plan_map(product_id_filter=plan.product_id),
"success_url": request.build_absolute_uri(reverse("success")),
"cancel_url": request.build_absolute_uri(reverse("cancel")),
}
if request.user.is_superuser:
cast["discounts"] = [{"coupon": settings.STRIPE_ADMIN_COUPON}]
session = stripe.checkout.Session.create(**cast)
Session.objects.create(user=request.user, session=session.id)
return redirect(session.url)
# return JsonResponse({'id': session.id})
except Exception as e:
# Raise a server error
return JsonResponse({"error": str(e)}, status=500)
class Cancel(LoginRequiredMixin, View):
def get(self, request, plan_name):
if not settings.STRIPE_ENABLED:
return redirect(reverse("home"))
plan = Plan.objects.get(name=plan_name)
try:
subscriptions = stripe.Subscription.list(
customer=request.user.stripe_id, price=plan.product_id
)
for subscription in subscriptions["data"]:
items = subscription["items"]["data"]
for item in items:
stripe.Subscription.delete(item["subscription"])
return render(request, "subscriptioncancel.html", {"plan": plan})
# return JsonResponse({'id': session.id})
except Exception as e:
# Raise a server error
logging.error(f"Error cancelling subscription for user: {e}")
return JsonResponse({"error": "True"}, status=500)
class Signup(CreateView):
form_class = NewUserForm
success_url = reverse_lazy("login")
success_url = reverse_lazy("two_factor:login")
template_name = "registration/signup.html"
def form_valid(self, form):
"""If the form is valid, save the associated model."""
self.object = form.save()
raw_sendmsg(
f"New user signup: {self.object.username} - {self.object.email}",
title="New user",
topic=settings.NOTIFY_TOPIC,
)
return super().form_valid(form)
def get(self, request, *args, **kwargs):
if not settings.REGISTRATION_OPEN:
return render(request, "registration/registration_closed.html")
super().get(request, *args, **kwargs)
class Portal(LoginRequiredMixin, View):
def get(self, request):
if not settings.STRIPE_ENABLED:
return redirect(reverse("home"))
session = stripe.billing_portal.Session.create(
customer=request.user.stripe_id,
return_url=request.build_absolute_uri(reverse("billing")),
)
return redirect(session.url)
return super().get(request, *args, **kwargs)

103
core/views/callbacks.py
View File

@@ -1,103 +0,0 @@
import logging
from datetime import datetime
import stripe
from django.conf import settings
from django.http import HttpResponse, JsonResponse
from django.views.decorators.csrf import csrf_exempt
from rest_framework.parsers import JSONParser
from rest_framework.views import APIView
from core.models import Plan, Session, User
logger = logging.getLogger(__name__)
class Callback(APIView):
parser_classes = [JSONParser]
@csrf_exempt
def post(self, request):
payload = request.body
sig_header = request.META["HTTP_STRIPE_SIGNATURE"]
try:
stripe.Webhook.construct_event(
payload, sig_header, settings.STRIPE_ENDPOINT_SECRET
)
except ValueError:
# Invalid payload
logger.error("Invalid payload")
return HttpResponse(status=400)
except stripe.error.SignatureVerificationError:
# Invalid signature
logger.error("Invalid signature")
return HttpResponse(status=400)
if request.data is None:
return JsonResponse({"success": False}, status=500)
if "type" in request.data.keys():
rtype = request.data["type"]
if rtype == "checkout.session.completed":
session = request.data["data"]["object"]["id"]
subscription_id = request.data["data"]["object"]["subscription"]
session_map = Session.objects.get(session=session)
if not session_map:
return JsonResponse({"success": False}, status=500)
user = session_map.user
session_map.subscription_id = subscription_id
session_map.save()
if rtype == "customer.subscription.updated":
stripe_id = request.data["data"]["object"]["customer"]
if not stripe_id:
logging.error("No stripe id")
return JsonResponse({"success": False}, status=500)
user = User.objects.get(stripe_id=stripe_id)
# ssubscription_active
subscription_id = request.data["data"]["object"]["id"]
sessions = Session.objects.filter(user=user)
session = None
for session_iter in sessions:
if session_iter.subscription_id == subscription_id:
session = session_iter
if not session:
logging.error(
f"No session found for subscription id {subscription_id}"
)
return JsonResponse({"success": False}, status=500)
# query Session objects
# iterate and check against product_id
session.request = request.data["request"]["id"]
product_id = request.data["data"]["object"]["plan"]["id"]
plan = Plan.objects.get(product_id=product_id)
if not plan:
logging.error(f"Plan not found: {product_id}")
return JsonResponse({"success": False}, status=500)
session.plan = plan
session.save()
elif rtype == "payment_intent.succeeded":
customer = request.data["data"]["object"]["customer"]
user = User.objects.get(stripe_id=customer)
if not user:
logging.error(f"No user found for customer: {customer}")
return JsonResponse({"success": False}, status=500)
session = Session.objects.get(request=request.data["request"]["id"])
user.plans.add(session.plan)
user.last_payment = datetime.utcnow()
user.save()
elif rtype == "customer.subscription.deleted":
customer = request.data["data"]["object"]["customer"]
user = User.objects.get(stripe_id=customer)
if not user:
logging.error(f"No user found for customer {customer}")
return JsonResponse({"success": False}, status=500)
product_id = request.data["data"]["object"]["plan"]["id"]
plan = Plan.objects.get(product_id=product_id)
user.plans.remove(plan)
user.save()
else:
return JsonResponse({"success": False}, status=500)
return JsonResponse({"success": True})

26
core/views/demo.py
View File

@@ -1,26 +0,0 @@
import uuid
from django.shortcuts import render
from django.views import View
class DemoModal(View):
template_name = "modals/modal.html"
def get(self, request):
return render(request, self.template_name)
class DemoWidget(View):
template_name = "widgets/widget.html"
def get(self, request):
unique = str(uuid.uuid4())[:8]
return render(request, self.template_name, {"unique": unique})
class DemoWindow(View):
template_name = "windows/window.html"
def get(self, request):
return render(request, self.template_name)

30
core/views/notifications.py Normal file
View File

@@ -0,0 +1,30 @@
from django.contrib.auth.mixins import LoginRequiredMixin
from mixins.views import ObjectUpdate
from core.forms import NotificationSettingsForm
from core.models import NotificationSettings
# Notifications - we create a new notification settings object if there isn't one
# Hence, there is only an update view, not a create view.
class NotificationsUpdate(LoginRequiredMixin, ObjectUpdate):
model = NotificationSettings
form_class = NotificationSettingsForm
page_title = "Update your notification settings"
page_subtitle = (
"At least the topic must be set if you want to receive notifications."
)
submit_url_name = "notifications_update"
submit_url_args = ["type"]
pk_required = False
hide_cancel = True
def get_object(self, **kwargs):
notification_settings, _ = NotificationSettings.objects.get_or_create(
user=self.request.user
)
return notification_settings

0
core/views/signal.py Normal file
View File