Add token based authentication feature

REQUIREMENTS.txt

@@ -53,5 +53,7 @@ django-bootstrap-pagination==1.7.1
 django-sortable-listview==0.43
 django-user-map
 djangorestframework==3.12.2
+pyjwt==1.7.1
+djangorestframework-simplejwt==4.4 
 django-rest-auth==0.9.5
 drf-yasg

dockerize/docker/REQUIREMENTS.txt

@@ -43,6 +43,9 @@ requests==2.23.0
 markdown==3.2.1
 
 djangorestframework==3.11.2
+pyjwt==1.7.1
+djangorestframework-simplejwt==4.4 
+
 sorl-thumbnail-serializer-field==0.2.1
 django-rest-auth==0.9.5
 drf-yasg==1.17.1

qgis-app/plugins/middleware.py

@@ -2,8 +2,7 @@
 # Author: A. Pasotti
 
 from django.contrib import auth
-from django.contrib.auth.models import User
-
+from rest_framework_simplejwt.authentication import JWTAuthentication
 
 def HttpAuthMiddleware(get_response):
     """
@@ -15,19 +14,28 @@ def middleware(request):
         if auth_basic:
             import base64
 
-            username, dummy, password = base64.decodestring(
-                auth_basic[6:].encode("utf8")
-            ).partition(b":")
-            username = username.decode("utf8")
-            password = password.decode("utf8")
-
-            user = auth.authenticate(username=username, password=password)
-            if user:
-                # User is valid.  Set request.user and persist user in the session
-                # by logging the user in.
-                request.user = user
-                auth.login(request, user)
-
+            if str(auth_basic).startswith('Bearer'):
+                # Validate JWT token, get the user and login
+                authentication = JWTAuthentication()
+                validated_token = authentication.get_validated_token(auth_basic[7:])
+                user = authentication.get_user(validated_token)
+                if user:
+                    request.user = user
+                    auth.login(request, user, backend='django.contrib.auth.backends.ModelBackend')
+
+            else:
+                username, dummy, password = base64.decodestring(
+                    auth_basic[6:].encode("utf8")
+                ).partition(b":")
+                username = username.decode("utf8")
+                password = password.decode("utf8")
+
+                user = auth.authenticate(username=username, password=password)
+                if user:
+                    # User is valid.  Set request.user and persist user in the session
+                    # by logging the user in.
+                    request.user = user
+                    auth.login(request, user)
         response = get_response(request)
 
         # Code to be executed for each request/response after

qgis-app/plugins/tests/test_token_auth.py

@@ -0,0 +1,116 @@
+import os
+from unittest.mock import patch
+
+from django.urls import reverse
+from django.test import Client, TestCase, override_settings
+from django.contrib.auth.models import User
+from django.core.files.uploadedfile import SimpleUploadedFile
+from plugins.models import Plugin, PluginVersion
+from plugins.forms import PackageUploadForm
+
+def do_nothing(*args, **kwargs):
+    pass
+
+TESTFILE_DIR = os.path.abspath(os.path.join(os.path.dirname(__file__), "testfiles"))
+
+class UploadWithTokenTestCase(TestCase):
+    fixtures = [
+        "fixtures/styles.json",
+        "fixtures/auth.json",
+        "fixtures/simplemenu.json",
+    ]
+
+    @override_settings(MEDIA_ROOT="api/tests")
+    def setUp(self):
+        self.client = Client()
+        self.url = reverse('plugin_upload')
+
+        # Create a test user
+        self.user = User.objects.create_user(
+            username='testuser',
+            password='testpassword',
+            email='[email protected]'
+        )
+
+    def test_upload_with_token(self):
+        create_token_url = reverse('token_obtain_pair')
+        data = {
+            'username': 'testuser',
+            'password': 'testpassword'
+        }
+
+        # Test POST request to create token
+        response = self.client.post(create_token_url, data)
+        self.assertEqual(response.status_code, 200)
+        self.assertTrue('access' in response.json())
+        self.assertTrue('refresh' in response.json())
+
+        access_token = response.json()['access']
+
+        valid_plugin = os.path.join(TESTFILE_DIR, "valid_plugin.zip_")
+        with open(valid_plugin, "rb") as file:
+            uploaded_file = SimpleUploadedFile(
+                "valid_plugin.zip_", file.read(),
+                content_type="application/zip")
+
+        c = Client(HTTP_AUTHORIZATION=f"Bearer {access_token}")
+        # Test POST request with access token
+        response = c.post(self.url, {
+            'package': uploaded_file,
+        })
+
+
+        self.assertEqual(response.status_code, 302)
+        self.assertTrue(Plugin.objects.filter(name='Test Plugin').exists())
+        self.assertEqual(
+            Plugin.objects.get(name='Test Plugin').tags.filter(
+                name__in=['python', 'example', 'test']).count(),
+            3)
+        self.assertTrue(PluginVersion.objects.filter(plugin__name='Test Plugin', version='0.0.1').exists())
+
+    def test_refresh_token(self):
+        create_token_url = reverse('token_obtain_pair')
+        data = {
+            'username': 'testuser',
+            'password': 'testpassword'
+        }
+
+        # Test POST request to create token
+        response = self.client.post(create_token_url, data)
+        refresh_token = response.json()['refresh']
+
+        refresh_token_url = reverse('token_refresh')
+
+        # Test POST request to create token
+        refresh_data = {
+            'refresh': refresh_token
+        }
+        response = self.client.post(refresh_token_url, refresh_data)
+        self.assertEqual(response.status_code, 200)
+        self.assertTrue('access' in response.json())
+
+        access_token = response.json()['access']
+
+        valid_plugin = os.path.join(TESTFILE_DIR, "valid_plugin.zip_")
+        with open(valid_plugin, "rb") as file:
+            uploaded_file = SimpleUploadedFile(
+                "valid_plugin.zip_", file.read(),
+                content_type="application/zip")
+
+        c = Client(HTTP_AUTHORIZATION=f"Bearer {access_token}")
+        # Test POST request with access token
+        response = c.post(self.url, {
+            'package': uploaded_file,
+        })
+
+
+        self.assertEqual(response.status_code, 302)
+        self.assertTrue(Plugin.objects.filter(name='Test Plugin').exists())
+        self.assertEqual(
+            Plugin.objects.get(name='Test Plugin').tags.filter(
+                name__in=['python', 'example', 'test']).count(),
+            3)
+        self.assertTrue(PluginVersion.objects.filter(plugin__name='Test Plugin', version='0.0.1').exists())
+
+
+

qgis-app/settings_docker.py

@@ -4,6 +4,7 @@
 from settings import *
 
 SITE_ROOT = os.path.dirname(os.path.realpath(__file__))
+from datetime import timedelta
 
 DEBUG = ast.literal_eval(os.environ.get("DEBUG", "True"))
 THUMBNAIL_DEBUG = DEBUG
@@ -63,6 +64,8 @@
     "feedjack",
     "preferences",
     "rest_framework",
+    'rest_framework.authtoken',
+    'rest_framework_simplejwt',
     "sorl_thumbnail_serializer",  # serialize image
     "drf_multiple_model",
     "drf_yasg",
@@ -120,3 +123,9 @@
 REST_FRAMEWORK = {
     "TEST_REQUEST_DEFAULT_FORMAT": "json",
 }
+
+# Token access and refresh validity
+SIMPLE_JWT = {
+    'ACCESS_TOKEN_LIFETIME': timedelta(days=15),
+    'REFRESH_TOKEN_LIFETIME': timedelta(days=180),
+}

qgis-app/urls.py

@@ -15,6 +15,10 @@
 # from users.views import *
 from homepage import homepage
 from rest_framework import permissions
+from rest_framework_simplejwt.views import (
+    TokenObtainPairView,
+    TokenRefreshView,
+)
 
 admin.autodiscover()
 
@@ -112,6 +116,11 @@
         url(r"^__debug__/", include(debug_toolbar.urls)),
     ]
 
+# Token
+urlpatterns += [
+    path('api/token/', TokenObtainPairView.as_view(), name='token_obtain_pair'),
+    path('api/token/refresh/', TokenRefreshView.as_view(), name='token_refresh'),
+]
 
 simplemenu.register(
     "/admin/",

readme.md

@@ -31,6 +31,32 @@ To update QGIS versions, go to **[Admin](https://plugins.qgis.org/admin/)** -> *
 This application is based on Django, written in python and deployed on the server using
 docker and rancher.
 
+## Token based authentication
+
+Users have the ability to generate a Simple JWT token by providing their credentials, which can then be utilized to access endpoints requiring authentication. Detailed guidance on the utilization of Simple JWT is provided in the official [documentation](https://django-rest-framework-simplejwt.readthedocs.io/en/latest/getting_started.html#usage).
+
+The endpoints are: 
+- Create a token: `https://plugins.qgis.org/api/token/`
+- Refresh token: `https://plugins.qgis.org/api/token/refresh`
+
+Examples:
+
+```sh
+# Create a token
+curl \
+  -X POST \
+  -H "Content-Type: application/json" \
+  -d '{"username": "yourusername", "password": "yourpassword"}' \
+  https://plugins.qgis.org/api/token/
+```
+
+```sh
+# Use the returned access token with the upload plugin endpoint
+curl \
+  -H "Authorization: Bearer the_access_token" \
+  https://plugins.qgis.org/plugins/add/
+```
+
 
 ## Contributing