From f3f0304ee41dc4a9ba0535ac47f9915c538e4605 Mon Sep 17 00:00:00 2001
From: Lova Andriarimalala <43842786+Xpirix@users.noreply.github.com>
Date: Mon, 18 Nov 2024 11:45:38 +0300
Subject: [PATCH] Fix Hub API permissions issues
---
.vscode/settings.json | 11 +
qgis-app/api/permissions.py | 35 +++
qgis-app/api/tests/test_resources_api.py | 267 ++++++++++++++++++++++-
qgis-app/api/views.py | 12 +-
qgis-app/templates/layouts/footer.html | 2 +-
5 files changed, 316 insertions(+), 11 deletions(-)
create mode 100644 .vscode/settings.json
diff --git a/.vscode/settings.json b/.vscode/settings.json
new file mode 100644
index 00000000..311cd262
--- /dev/null
+++ b/.vscode/settings.json
@@ -0,0 +1,11 @@
+{
+ "python.testing.unittestArgs": [
+ "-v",
+ "-s",
+ "./qgis-app",
+ "-p",
+ "test_*.py"
+ ],
+ "python.testing.pytestEnabled": false,
+ "python.testing.unittestEnabled": true
+}
\ No newline at end of file
diff --git a/qgis-app/api/permissions.py b/qgis-app/api/permissions.py
index 58765bf5..491634ce 100644
--- a/qgis-app/api/permissions.py
+++ b/qgis-app/api/permissions.py
@@ -1,4 +1,11 @@
from rest_framework import permissions
+from rest_framework.permissions import BasePermission
+from rest_framework_simplejwt.authentication import JWTAuthentication
+from django.contrib.auth.models import User
+from rest_framework_simplejwt.exceptions import InvalidToken, TokenError
+from rest_framework_simplejwt.token_blacklist.models import BlacklistedToken, OutstandingToken
+import datetime
+from api.models import UserOutstandingToken
MANAGER_GROUP = "Style Managers"
@@ -21,3 +28,31 @@ def has_object_permission(self, request, view, obj):
is_manager = user.groups.filter(name=MANAGER_GROUP).exists()
return user == obj.creator or user.is_staff or is_manager
+
+class HasValidToken(BasePermission):
+ def has_permission(self, request, view):
+ auth_token = request.META.get("HTTP_AUTHORIZATION")
+ if not str(auth_token).startswith('Bearer'):
+ return False
+
+ # Validate JWT token
+ authentication = JWTAuthentication()
+ try:
+ validated_token = authentication.get_validated_token(auth_token[7:])
+ user_id = validated_token.payload.get('user_id')
+ jti = validated_token.payload.get('refresh_jti')
+ token_id = OutstandingToken.objects.get(jti=jti).pk
+ is_blacklisted = BlacklistedToken.objects.filter(token_id=token_id).exists()
+ if not user_id or is_blacklisted:
+ return False
+
+ user = User.objects.get(pk=user_id)
+ if not user:
+ return False
+ user_token = UserOutstandingToken.objects.get(token__pk=token_id, user=user)
+ user_token.last_used_on = datetime.datetime.now()
+ user_token.save()
+ request.user_token = user_token
+ return True
+ except (InvalidToken, TokenError):
+ return False
\ No newline at end of file
diff --git a/qgis-app/api/tests/test_resources_api.py b/qgis-app/api/tests/test_resources_api.py
index 45c6e5d5..c529a592 100644
--- a/qgis-app/api/tests/test_resources_api.py
+++ b/qgis-app/api/tests/test_resources_api.py
@@ -15,6 +15,9 @@
from django.core.files.uploadedfile import SimpleUploadedFile
from wavefronts.models import Wavefront
+from rest_framework_simplejwt.token_blacklist.models import OutstandingToken
+from api.models import UserOutstandingToken
+
GPKG_DIR = join(dirname(dirname(dirname(__file__))), "geopackages", "tests", "gpkgfiles")
LAYERDEFINITION_DIR = join(dirname(dirname(dirname(__file__))), "layerdefinitions", "tests", "testfiles")
MODELS_DIR = join(dirname(dirname(dirname(__file__))), "models", "tests", "modelfiles")
@@ -50,8 +53,21 @@ def setUp(self):
super().setUp()
self.client = APIClient()
self.user = User.objects.create_user(username='testuser', password='testpass')
- self.token = RefreshToken.for_user(self.user)
- self.client.credentials(HTTP_AUTHORIZATION=f'Bearer {self.token.access_token}')
+ self.client.login(username='testuser', password='testpass')
+ self.refresh = RefreshToken.for_user(self.user)
+ self.outstanding_token = OutstandingToken.objects.get(jti=self.refresh['jti'])
+ self.user_token = UserOutstandingToken.objects.create(
+ user=self.user,
+ token=self.outstanding_token,
+ is_blacklisted=False,
+ is_newly_created=True
+ )
+ self.url = reverse('user_token_detail', args=[self.user_token.pk])
+ response = self.client.get(self.url)
+ self.assertEqual(response.status_code, 200)
+ self.client.logout()
+ access_token = response.context.get('access_token')
+ self.client.credentials(HTTP_AUTHORIZATION=f'Bearer {access_token}')
def test_create_geopackage(self):
url = reverse('resource-create')
@@ -202,6 +218,44 @@ def test_create_unsupported_resource_type(self):
self.assertEqual(response.status_code, 400)
self.assertEqual(response.data, {"resource_type": "Resource type not supported"})
+ def test_create_with_invalid_token(self):
+ url = reverse('resource-create')
+ self.client.credentials(HTTP_AUTHORIZATION='Bearer invalidtoken')
+ uploaded_thumbnail = SimpleUploadedFile(
+ self.thumbnail_content.name, self.thumbnail_content.read()
+ )
+ uploaded_gpkg = SimpleUploadedFile(
+ self.gpkg_file_content.name, self.gpkg_file_content.read()
+ )
+ data = {
+ "resource_type": "geopackage",
+ "name": "Test Geopackage",
+ "description": "A test geopackage",
+ "thumbnail_full": uploaded_thumbnail,
+ "file": uploaded_gpkg,
+ }
+ response = self.client.post(url, data, format='multipart')
+ self.assertEqual(response.status_code, 401)
+
+ def test_create_with_blacklisted_token(self):
+ url = reverse('resource-create')
+ self.refresh.blacklist()
+ uploaded_thumbnail = SimpleUploadedFile(
+ self.thumbnail_content.name, self.thumbnail_content.read()
+ )
+ uploaded_gpkg = SimpleUploadedFile(
+ self.gpkg_file_content.name, self.gpkg_file_content.read()
+ )
+ data = {
+ "resource_type": "geopackage",
+ "name": "Test Geopackage",
+ "description": "A test geopackage",
+ "thumbnail_full": uploaded_thumbnail,
+ "file": uploaded_gpkg,
+ }
+ response = self.client.post(url, data, format='multipart')
+ self.assertEqual(response.status_code, 403)
+
@override_settings(MEDIA_ROOT=tempfile.mkdtemp())
class TestResourceDetailView(SetUpTest, TestCase):
@@ -210,8 +264,21 @@ def setUp(self):
super().setUp()
self.client = APIClient()
self.user = User.objects.create_user(username='testuser', password='testpass')
- self.token = RefreshToken.for_user(self.user)
- self.client.credentials(HTTP_AUTHORIZATION=f'Bearer {self.token.access_token}')
+ self.client.login(username='testuser', password='testpass')
+ self.refresh = RefreshToken.for_user(self.user)
+ self.outstanding_token = OutstandingToken.objects.get(jti=self.refresh['jti'])
+ self.user_token = UserOutstandingToken.objects.create(
+ user=self.user,
+ token=self.outstanding_token,
+ is_blacklisted=False,
+ is_newly_created=True
+ )
+ self.url = reverse('user_token_detail', args=[self.user_token.pk])
+ response = self.client.get(self.url)
+ self.assertEqual(response.status_code, 200)
+ self.client.logout()
+ access_token = response.context.get('access_token')
+ self.client.credentials(HTTP_AUTHORIZATION=f'Bearer {access_token}')
uploaded_thumbnail = SimpleUploadedFile(
self.thumbnail_content.name, self.thumbnail_content.read()
@@ -406,4 +473,194 @@ def test_delete_wavefront(self):
url = reverse("resource-detail", kwargs={"uuid": self.wavefront.uuid, "resource_type": "3dmodel"})
response = self.client.delete(url)
self.assertEqual(response.status_code, 204)
- self.assertFalse(Wavefront.objects.filter(uuid=self.wavefront.uuid).exists())
\ No newline at end of file
+ self.assertFalse(Wavefront.objects.filter(uuid=self.wavefront.uuid).exists())
+
+ def test_update_geopackage_with_invalid_token(self):
+ url = reverse("resource-detail", kwargs={"uuid": self.geopackage.uuid, "resource_type": "geopackage"})
+ self.client.credentials(HTTP_AUTHORIZATION='Bearer invalidtoken')
+ data = {
+ "name": "Updated Geopackage",
+ "description": "Updated description",
+ "thumbnail_full": self.geopackage.thumbnail_image,
+ "file": self.geopackage.file,
+ }
+ response = self.client.put(url, data, format="multipart")
+ self.assertEqual(response.status_code, 401)
+
+ def test_update_geopackage_with_blacklisted_token(self):
+ url = reverse("resource-detail", kwargs={"uuid": self.geopackage.uuid, "resource_type": "geopackage"})
+ self.refresh.blacklist()
+ data = {
+ "name": "Updated Geopackage",
+ "description": "Updated description",
+ "thumbnail_full": self.geopackage.thumbnail_image,
+ "file": self.geopackage.file,
+ }
+ response = self.client.put(url, data, format="multipart")
+ self.assertEqual(response.status_code, 403)
+
+ def test_update_layerdefinition_with_invalid_token(self):
+ url = reverse("resource-detail", kwargs={"uuid": self.layerdefinition.uuid, "resource_type": "layerdefinition"})
+ self.client.credentials(HTTP_AUTHORIZATION='Bearer invalidtoken')
+ data = {
+ "name": "Updated Layer Definition",
+ "description": "Updated description",
+ "thumbnail_full": self.layerdefinition.thumbnail_image,
+ "file": self.layerdefinition.file,
+ }
+ response = self.client.put(url, data, format="multipart")
+ self.assertEqual(response.status_code, 401)
+
+ def test_update_layerdefinition_with_blacklisted_token(self):
+ url = reverse("resource-detail", kwargs={"uuid": self.layerdefinition.uuid, "resource_type": "layerdefinition"})
+ self.refresh.blacklist()
+ data = {
+ "name": "Updated Layer Definition",
+ "description": "Updated description",
+ "thumbnail_full": self.layerdefinition.thumbnail_image,
+ "file": self.layerdefinition.file,
+ }
+ response = self.client.put(url, data, format="multipart")
+ self.assertEqual(response.status_code, 403)
+
+ def test_update_model_with_invalid_token(self):
+ url = reverse("resource-detail", kwargs={"uuid": self.model.uuid, "resource_type": "model"})
+ self.client.credentials(HTTP_AUTHORIZATION='Bearer invalidtoken')
+ data = {
+ "name": "Updated Model",
+ "description": "Updated description",
+ "thumbnail_full": self.model.thumbnail_image,
+ "file": self.model.file,
+ }
+ response = self.client.put(url, data, format="multipart")
+ self.assertEqual(response.status_code, 401)
+
+ def test_update_model_with_blacklisted_token(self):
+ url = reverse("resource-detail", kwargs={"uuid": self.model.uuid, "resource_type": "model"})
+ self.refresh.blacklist()
+ data = {
+ "name": "Updated Model",
+ "description": "Updated description",
+ "thumbnail_full": self.model.thumbnail_image,
+ "file": self.model.file,
+ }
+ response = self.client.put(url, data, format="multipart")
+ self.assertEqual(response.status_code, 403)
+
+ def test_update_style_with_invalid_token(self):
+ url = reverse("resource-detail", kwargs={"uuid": self.style.uuid, "resource_type": "style"})
+ self.client.credentials(HTTP_AUTHORIZATION='Bearer invalidtoken')
+ data = {
+ "name": "Updated Style",
+ "description": "Updated description",
+ "thumbnail_full": self.style.thumbnail_image,
+ "file": self.style.file,
+ }
+ response = self.client.put(url, data, format="multipart")
+ self.assertEqual(response.status_code, 401)
+
+ def test_update_style_with_blacklisted_token(self):
+ url = reverse("resource-detail", kwargs={"uuid": self.style.uuid, "resource_type": "style"})
+ self.refresh.blacklist()
+ data = {
+ "name": "Updated Style",
+ "description": "Updated description",
+ "thumbnail_full": self.style.thumbnail_image,
+ "file": self.style.file,
+ }
+ response = self.client.put(url, data, format="multipart")
+ self.assertEqual(response.status_code, 403)
+
+ def test_update_wavefront_with_invalid_token(self):
+ url = reverse("resource-detail", kwargs={"uuid": self.wavefront.uuid, "resource_type": "3dmodel"})
+ self.client.credentials(HTTP_AUTHORIZATION='Bearer invalidtoken')
+ data = {
+ "name": "Updated 3D Model",
+ "description": "Updated description",
+ "thumbnail_full": self.wavefront.thumbnail_image,
+ "file": self.wavefront.file,
+ }
+ response = self.client.put(url, data, format="multipart")
+ self.assertEqual(response.status_code, 401)
+
+ def test_update_wavefront_with_blacklisted_token(self):
+ url = reverse("resource-detail", kwargs={"uuid": self.wavefront.uuid, "resource_type": "3dmodel"})
+ self.refresh.blacklist()
+ data = {
+ "name": "Updated 3D Model",
+ "description": "Updated description",
+ "thumbnail_full": self.wavefront.thumbnail_image,
+ "file": self.wavefront.file,
+ }
+ response = self.client.put(url, data, format="multipart")
+ self.assertEqual(response.status_code, 403)
+
+ def test_delete_geopackage_with_invalid_token(self):
+ url = reverse("resource-detail", kwargs={"uuid": self.geopackage.uuid, "resource_type": "geopackage"})
+ self.client.credentials(HTTP_AUTHORIZATION='Bearer invalidtoken')
+ response = self.client.delete(url)
+ self.assertEqual(response.status_code, 401)
+ self.assertTrue(Geopackage.objects.filter(uuid=self.geopackage.uuid).exists())
+
+ def test_delete_geopackage_with_blacklisted_token(self):
+ url = reverse("resource-detail", kwargs={"uuid": self.geopackage.uuid, "resource_type": "geopackage"})
+ self.refresh.blacklist()
+ response = self.client.delete(url)
+ self.assertEqual(response.status_code, 403)
+ self.assertTrue(Geopackage.objects.filter(uuid=self.geopackage.uuid).exists())
+
+ def test_delete_layerdefinition_with_invalid_token(self):
+ url = reverse("resource-detail", kwargs={"uuid": self.layerdefinition.uuid, "resource_type": "layerdefinition"})
+ self.client.credentials(HTTP_AUTHORIZATION='Bearer invalidtoken')
+ response = self.client.delete(url)
+ self.assertEqual(response.status_code, 401)
+ self.assertTrue(LayerDefinition.objects.filter(uuid=self.layerdefinition.uuid).exists())
+
+ def test_delete_layerdefinition_with_blacklisted_token(self):
+ url = reverse("resource-detail", kwargs={"uuid": self.layerdefinition.uuid, "resource_type": "layerdefinition"})
+ self.refresh.blacklist()
+ response = self.client.delete(url)
+ self.assertEqual(response.status_code, 403)
+ self.assertTrue(LayerDefinition.objects.filter(uuid=self.layerdefinition.uuid).exists())
+
+ def test_delete_model_with_invalid_token(self):
+ url = reverse("resource-detail", kwargs={"uuid": self.model.uuid, "resource_type": "model"})
+ self.client.credentials(HTTP_AUTHORIZATION='Bearer invalidtoken')
+ response = self.client.delete(url)
+ self.assertEqual(response.status_code, 401)
+ self.assertTrue(Model.objects.filter(uuid=self.model.uuid).exists())
+
+ def test_delete_model_with_blacklisted_token(self):
+ url = reverse("resource-detail", kwargs={"uuid": self.model.uuid, "resource_type": "model"})
+ self.refresh.blacklist()
+ response = self.client.delete(url)
+ self.assertEqual(response.status_code, 403)
+ self.assertTrue(Model.objects.filter(uuid=self.model.uuid).exists())
+
+ def test_delete_style_with_invalid_token(self):
+ url = reverse("resource-detail", kwargs={"uuid": self.style.uuid, "resource_type": "style"})
+ self.client.credentials(HTTP_AUTHORIZATION='Bearer invalidtoken')
+ response = self.client.delete(url)
+ self.assertEqual(response.status_code, 401)
+ self.assertTrue(Style.objects.filter(uuid=self.style.uuid).exists())
+
+ def test_delete_style_with_blacklisted_token(self):
+ url = reverse("resource-detail", kwargs={"uuid": self.style.uuid, "resource_type": "style"})
+ self.refresh.blacklist()
+ response = self.client.delete(url)
+ self.assertEqual(response.status_code, 403)
+ self.assertTrue(Style.objects.filter(uuid=self.style.uuid).exists())
+
+ def test_delete_wavefront_with_invalid_token(self):
+ url = reverse("resource-detail", kwargs={"uuid": self.wavefront.uuid, "resource_type": "3dmodel"})
+ self.client.credentials(HTTP_AUTHORIZATION='Bearer invalidtoken')
+ response = self.client.delete(url)
+ self.assertEqual(response.status_code, 401)
+ self.assertTrue(Wavefront.objects.filter(uuid=self.wavefront.uuid).exists())
+
+ def test_delete_wavefront_with_blacklisted_token(self):
+ url = reverse("resource-detail", kwargs={"uuid": self.wavefront.uuid, "resource_type": "3dmodel"})
+ self.refresh.blacklist()
+ response = self.client.delete(url)
+ self.assertEqual(response.status_code, 403)
+ self.assertTrue(Wavefront.objects.filter(uuid=self.wavefront.uuid).exists())
diff --git a/qgis-app/api/views.py b/qgis-app/api/views.py
index 829c6f94..b60b2a69 100644
--- a/qgis-app/api/views.py
+++ b/qgis-app/api/views.py
@@ -35,9 +35,12 @@
from wavefronts.models import Wavefront
from api.models import UserOutstandingToken
from rest_framework_simplejwt.token_blacklist.models import OutstandingToken
+
from rest_framework.response import Response
from rest_framework.parsers import MultiPartParser, FormParser
from rest_framework import status
+from django.contrib.auth.mixins import LoginRequiredMixin
+from api.permissions import HasValidToken
def filter_resource_type(queryset, request, *args, **kwargs):
resource_type = request.query_params["resource_type"]
@@ -189,7 +192,7 @@ def get(self, request, *args, **kwargs):
return response
-class UserTokenDetailView(DetailView):
+class UserTokenDetailView(LoginRequiredMixin, DetailView):
"""
Hub token detail
"""
@@ -235,7 +238,7 @@ def get_context_data(self, **kwargs):
return context
-class UserTokenListView(ListView):
+class UserTokenListView(LoginRequiredMixin, ListView):
"""
Hub token list
"""
@@ -283,7 +286,6 @@ def user_token_create(request):
@login_required
@transaction.atomic
def user_token_update(request, token_id):
- print(token_id)
user_token = get_object_or_404(
UserOutstandingToken,
pk=token_id,
@@ -379,7 +381,7 @@ class ResourceCreateView(APIView):
Create a new Resource
"""
authentication_classes = [JWTAuthentication]
- permission_classes = [IsAuthenticated]
+ permission_classes = [HasValidToken]
parser_classes = [MultiPartParser, FormParser]
def post(self, request, *args, **kwargs):
@@ -412,7 +414,7 @@ class ResourceDetailView(APIView):
Retrieve or update a Resource
"""
authentication_classes = [JWTAuthentication]
- permission_classes = [IsAuthenticated]
+ permission_classes = [HasValidToken]
def get(self, request, *args, **kwargs):
uuid = kwargs.get("uuid")
diff --git a/qgis-app/templates/layouts/footer.html b/qgis-app/templates/layouts/footer.html
index afd7fc4b..1c8f52ed 100644
--- a/qgis-app/templates/layouts/footer.html
+++ b/qgis-app/templates/layouts/footer.html
@@ -48,7 +48,7 @@