wg-manager/wg_connection_manager/views.py

from django.shortcuts import render, get_object_or_404, redirect
from django.http import HttpRequest, HttpResponseNotAllowed, HttpResponseNotFound, HttpResponseBadRequest
from .models import UserConnection
from django.contrib.auth.decorators import login_required
from .dj_wg_manager_factory import get_dj_wg_manager
from django import forms
from django_otp.decorators import otp_required


class UserConnectionForm(forms.ModelForm):
    class Meta:
        model = UserConnection
        fields = ["public_key", "preshared_key", "vpn_ip", "notes"]


@login_required
def list_connections(request: HttpRequest):
    user = request.user
    if not user.is_verified():
        return redirect("two_factor:setup")
    if request.method == "POST":
        form = UserConnectionForm(request.POST)
        if form.is_valid():
            c = UserConnection(
                user=user,
                preshared_key=form.cleaned_data["preshared_key"],
                public_key=form.cleaned_data["public_key"],
                vpn_ip=form.cleaned_data["vpn_ip"],
                notes=form.cleaned_data["notes"],
            )
            c.save()
            form = UserConnectionForm()
    else:
        form = UserConnectionForm()
    if user.is_superuser:
        connections = UserConnection.objects.filter().order_by("user__username")
    else:
        connections = UserConnection.objects.filter(user=user)

    return render(request, "connection_list.html", context={"connections": connections, "form": form})


@otp_required
def toggle_connection(request: HttpRequest, connection_id: int):

    if request.method != "POST" or "state" not in request.POST:
        return HttpResponseBadRequest()
    if request.POST["state"] not in ["enabled", "disabled"]:
        return HttpResponseBadRequest("invalid state")
    connection = get_object_or_404(UserConnection, pk=connection_id)
    if connection.user != request.user and not request.user.is_superuser:
        return HttpResponseNotFound()
    wg_manager = get_dj_wg_manager()
    match request.POST["state"]:
        case "enabled":
            wg_manager.add_peer(connection)
        case "disabled":
            wg_manager.remove_peer(connection)
    return redirect("connections_list")


# @login_required
@otp_required
def delete_connection(request: HttpRequest, connection_id: int):
    connection = get_object_or_404(UserConnection, pk=connection_id)
    if connection.user != request.user and not request.user.is_superuser:
        return HttpResponseNotFound()
    # sync and disable before delete
    wg_manager = get_dj_wg_manager()
    wg_manager.sync()
    if connection.active:
        wg_manager.remove_peer(connection)
    connection.delete()
    return redirect("connections_list")


@login_required
def sync(request: HttpRequest):
    if request.user.is_superuser:
        wg_manager = get_dj_wg_manager()
        wg_manager.sync()
        return redirect("connections_list")
    return HttpResponseNotAllowed()