83 lines
2.9 KiB
Python
83 lines
2.9 KiB
Python
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()
|