wireguard_webadmin integration

integration_manager/admin.py

@@ -0,0 +1,9 @@
+from django.contrib import admin
+from .models import ExternalIntegration
+
+class ExternalIntegrationAdmin(admin.ModelAdmin):
+    list_display = ('id', 'name', 'integration_type', 'integration_url', 'wireguard_webadmin_default_user_level', 'token', 'created_at', 'updated_at')
+    search_fields = ('name', 'integration_type', 'integration_url', 'wireguard_webadmin_default_user_level', 'token', 'created_at', 'updated_at')
+    readonly_fields = ('created_at', 'updated_at')
+
+admin.site.register(ExternalIntegration, ExternalIntegrationAdmin)

integration_manager/apps.py

@@ -0,0 +1,6 @@
+from django.apps import AppConfig
+
+
+class IntegrationManagerConfig(AppConfig):
+    default_auto_field = 'django.db.models.BigAutoField'
+    name = 'integration_manager'

integration_manager/forms.py

@@ -0,0 +1,88 @@
+from django import forms
+from crispy_forms.helper import FormHelper
+from crispy_forms.layout import Layout, Submit, Row, Column, HTML
+from .models import ExternalIntegration
+import requests
+
+
+class WireGuardWebAdminForm(forms.ModelForm):
+    token = forms.CharField(widget=forms.PasswordInput, required=False)
+
+    class Meta:
+        model = ExternalIntegration
+        fields = ['integration_url', 'wireguard_webadmin_default_user_level', 'token']
+
+    def __init__(self, *args, **kwargs):
+        self.user = kwargs.pop('user', None)
+        super(WireGuardWebAdminForm, self).__init__(*args, **kwargs)
+        self.helper = FormHelper()
+        self.helper.form_method = 'post'
+        if self.instance.pk:
+            delete_html = "<a href='javascript:void(0)' class='btn btn-outline-danger' data-command='delete' onclick='openCommandDialog(this)'>Delete</a>"
+            if self.instance.token:
+                self.fields['token'].widget.attrs['placeholder'] = '************'
+        else:
+            delete_html = ''
+        self.helper.layout = Layout(
+            Row(
+                Column('integration_url', css_class='col-md-12'),
+            ),
+            Row(
+                Column('wireguard_webadmin_default_user_level', css_class='col-md-12'),
+            ),
+            Row(
+                Column('token', css_class='col-md-12'),
+            ),
+            Row(
+                Column(
+                    Submit('submit', 'Salvar', css_class='btn btn-success'),
+                    HTML(' <a class="btn btn-secondary" href="/wireguard_webadmin/">Back</a> '),
+                    HTML(delete_html),
+                    css_class='col-md-12'),
+                css_class='form-row'
+            )
+        )
+
+    def clean(self):
+        cleaned_data = super().clean()
+        integration_url = cleaned_data.get('integration_url')
+        wireguard_webadmin_default_user_level = cleaned_data.get('wireguard_webadmin_default_user_level')
+        token = cleaned_data.get('token')
+        if integration_url.endswith('/'):
+            cleaned_data['integration_url'] = integration_url[:-1]
+        if not token and self.instance.token:
+            cleaned_data['token'] = self.instance.token
+
+        if not integration_url.startswith('https://'):
+            raise forms.ValidationError('Please use https://')
+
+        api_test_url = f"{cleaned_data['integration_url']}/api/routerfleet_get_user_token/"
+        api_test_url += f"?key={cleaned_data['token']}"
+        api_test_url += f"&username={self.user.username}&action=test"
+
+        try:
+            api_test = requests.get(api_test_url)
+        except:
+            raise forms.ValidationError('Error connecting to API')
+
+        try:
+            if api_test.status_code == 403:
+                api_response = {}
+            else:
+                api_response = api_test.json()
+        except:
+            raise forms.ValidationError('Error parsing API response')
+
+        if api_test.status_code == 403:
+            raise forms.ValidationError('Invalid token')
+        elif api_test.status_code == 400:
+            if api_response.get('message'):
+                raise forms.ValidationError(api_response.get('message'))
+            else:
+                raise forms.ValidationError(f'Error authenticating with API. Status Code: {api_test.status_code}')
+        elif api_test.status_code != 200:
+            raise forms.ValidationError(f'Error connecting to API. Status Code: {api_test.status_code}')
+
+        return cleaned_data
+
+

integration_manager/migrations/0001_initial.py

@@ -0,0 +1,27 @@
+# Generated by Django 5.0.3 on 2024-04-03 19:21
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    initial = True
+
+    dependencies = [
+    ]
+
+    operations = [
+        migrations.CreateModel(
+            name='ExternalIntegration',
+            fields=[
+                ('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
+                ('name', models.CharField(max_length=100, unique=True)),
+                ('integration_type', models.CharField(choices=[('wireguard_webadmin', 'WireGuard WebAdmin')], max_length=100)),
+                ('integration_url', models.URLField()),
+                ('wireguard_webadmin_default_user_level', models.PositiveIntegerField(choices=[(10, 'Debugging Analyst'), (20, 'View Only User'), (30, 'Peer Manager'), (40, 'Manager'), (50, 'Administrator')], default=0)),
+                ('token', models.CharField(max_length=100)),
+                ('created_at', models.DateTimeField(auto_now_add=True)),
+                ('updated_at', models.DateTimeField(auto_now=True)),
+            ],
+        ),
+    ]

integration_manager/models.py

@@ -0,0 +1,14 @@
+from django.db import models
+
+
+class ExternalIntegration(models.Model):
+    name = models.CharField(max_length=100, unique=True)
+    integration_type = models.CharField(max_length=100, choices=(('wireguard_webadmin', 'WireGuard WebAdmin'), ))
+    integration_url = models.URLField()
+    wireguard_webadmin_default_user_level = models.PositiveIntegerField(default=0, choices=((0, 'Do not create users'), (10, 'Debugging Analyst'), (20, 'View Only User'), (30, 'Peer Manager'), (40, 'Manager'), (50, 'Administrator')))
+    token = models.CharField(max_length=100)
+    created_at = models.DateTimeField(auto_now_add=True)
+    updated_at = models.DateTimeField(auto_now=True)
+
+    def __str__(self):
+        return self.name

integration_manager/tests.py

@@ -0,0 +1,3 @@
+from django.test import TestCase
+
+# Create your tests here.

integration_manager/views.py

@@ -0,0 +1,76 @@
+from django.contrib.auth.decorators import login_required
+from django.shortcuts import render, redirect, get_object_or_404
+from .models import ExternalIntegration
+from .forms import WireGuardWebAdminForm
+from django.contrib import messages
+import requests
+
+
+@login_required()
+def view_wireguard_webadmin_launcher(request):
+    context = {
+        'page_title': 'WireGuard WebAdmin Launcher',
+        'wireguard_integration': ExternalIntegration.objects.filter(name='wireguard_webadmin', integration_type='wireguard_webadmin').first()
+    }
+    return render(request, 'integration_manager/wireguard_webadmin_launcher.html', context=context)
+
+
+@login_required()
+def view_launch_wireguard_webadmin(request):
+    wireguard_integration = get_object_or_404(ExternalIntegration, name='wireguard_webadmin', integration_type='wireguard_webadmin')
+    api_url = f"{wireguard_integration.integration_url}/api/routerfleet_get_user_token/"
+    api_url += f"?key={wireguard_integration.token}"
+    api_url += f"&username={request.user.username}&action=login"
+    api_url += f"&default_user_level={wireguard_integration.wireguard_webadmin_default_user_level}"
+    try:
+        api_response = requests.get(api_url)
+    except:
+        messages.warning(request, 'Error connecting to API')
+        return redirect('/wireguard_webadmin/')
+
+    try:
+        if api_response.status_code == 200:
+            api_json = api_response.json()
+        else:
+            api_json = {}
+    except:
+        messages.warning(request, 'Error parsing API response')
+        return redirect('/wireguard_webadmin/')
+
+    if api_response.status_code == 200:
+        redirect_url = f"{wireguard_integration.integration_url}/accounts/routerfleet_authenticate_session/"
+        redirect_url += f"?token={api_json.get('authentication_token')}"
+        return redirect(redirect_url)
+    else:
+        messages.warning(request, f'Error authenticating with API. Status Code: {api_response.status_code}')
+        return redirect('/wireguard_webadmin/')
+
+
+@login_required()
+def view_manage_wireguard_integration(request):
+    context = {
+        'page_title': 'Manage WireGuard Integration',
+        'delete_confirmation_message': 'Are you sure you want to delete this integration? This action cannot be undone. Type delete in the box below to confirm.'
+    }
+    wireguard_integration = ExternalIntegration.objects.filter(name='wireguard_webadmin', integration_type='wireguard_webadmin').first()
+    if request.GET.get('action') == 'delete':
+        if request.GET.get('confirmation') == 'delete':
+            wireguard_integration.delete()
+            messages.success(request, 'WireGuard WebAdmin integration deleted')
+            return redirect('/wireguard_webadmin/')
+        else:
+            messages.warning(request, 'Invalid confirmation. Integration not deleted')
+            return redirect('/wireguard_webadmin/')
+
+    form = WireGuardWebAdminForm(request.POST or None, instance=wireguard_integration, user=request.user)
+
+    if form.is_valid():
+        this_form = form.save(commit=False)
+        this_form.name = 'wireguard_webadmin'
+        this_form.integration_type = 'wireguard_webadmin'
+        this_form.save()
+        messages.success(request, 'WireGuard WebAdmin integration saved')
+        return redirect('/wireguard_webadmin/')
+
+    context['form'] = form
+    return render(request, 'generic_form.html', context=context)