Fix for plugin auto-approval logic

Author: Xpirix

qgis-app/plugins/api.py

@@ -120,7 +120,7 @@ def plugin_upload(package, **kwargs):
                 package.len,
                 "UTF-8",
             ),
-            "approved": request.user.has_perm("plugins.can_approve") or plugin.approved,
+            "approved": request.user.has_perm("plugins.can_approve"),
         }
 
         # Optional version metadata

qgis-app/plugins/tests/test_plugin_create_empty.py

@@ -291,5 +291,90 @@ def test_create_empty_plugin_user_without_full_name(self):
         # Should use username as author
         self.assertEqual(plugin.author, "noname")
 
+    @patch("plugins.views.plugin_notify", new=do_nothing)
+    @patch("plugins.tasks.generate_plugins_xml", new=do_nothing)
+    @patch("plugins.validator._check_url_link", new=do_nothing)
+    @patch("plugins.security_utils.run_security_scan", new=do_nothing)
+    def test_version_create_not_auto_approved_for_untrusted_user_on_approved_plugin(self):
+        """
+        Security: uploading a new version via the web form must NOT be auto-approved
+        because the plugin already has an approved version. Only the user's
+        can_approve permission should grant approval.
+        """
+        self.client.login(username="testuser", password="testpassword")
+
+        # Create empty plugin then upload first version
+        create_data = {"package_name": "test_modul", "name": "Test Module Plugin"}
+        self.client.post(self.url, create_data)
+        plugin = Plugin.objects.get(package_name="test_modul")
+
+        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"
+            )
+        upload_url = reverse("version_create", args=[plugin.package_name])
+        self.client.post(upload_url, {"package": uploaded_file})
+
+        # Simulate staff approval of the first version
+        first_version = plugin.pluginversion_set.get(version="0.0.1")
+        first_version.approved = True
+        first_version.save()
+
+        # Upload a second version as the same untrusted user
+        valid_plugin_v2 = os.path.join(TESTFILE_DIR, "valid_plugin_0.0.2.zip_")
+        with open(valid_plugin_v2, "rb") as file:
+            uploaded_file_v2 = SimpleUploadedFile(
+                "valid_plugin_0.0.2.zip_", file.read(), content_type="application/zip"
+            )
+        self.client.post(upload_url, {"package": uploaded_file_v2})
+
+        second_version = plugin.pluginversion_set.get(version="0.0.2")
+        self.assertFalse(
+            second_version.approved,
+            "New version must NOT be auto-approved just because the plugin is approved; "
+            "only user trust (can_approve) should grant approval.",
+        )
+
+    @patch("plugins.views.plugin_notify", new=do_nothing)
+    @patch("plugins.tasks.generate_plugins_xml", new=do_nothing)
+    @patch("plugins.validator._check_url_link", new=do_nothing)
+    @patch("plugins.security_utils.run_security_scan", new=do_nothing)
+    def test_version_create_auto_approved_for_trusted_user(self):
+        """
+        A user with can_approve permission should have their uploaded version
+        approved automatically via the version_create view.
+        """
+        from django.contrib.auth.models import Permission
+        from django.contrib.contenttypes.models import ContentType
+        from plugins.models import Plugin as PluginModel
+
+        ct = ContentType.objects.get_for_model(PluginModel)
+        perm = Permission.objects.get(codename="can_approve", content_type=ct)
+        self.user.user_permissions.add(perm)
+        # Refresh to bust Django's permission cache
+        self.user = self.user.__class__.objects.get(pk=self.user.pk)
+
+        self.client.login(username="testuser", password="testpassword")
+
+        # Create empty plugin then upload first version
+        create_data = {"package_name": "test_modul", "name": "Test Module Plugin"}
+        self.client.post(self.url, create_data)
+        plugin = Plugin.objects.get(package_name="test_modul")
+
+        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"
+            )
+        upload_url = reverse("version_create", args=[plugin.package_name])
+        self.client.post(upload_url, {"package": uploaded_file})
+
+        version = plugin.pluginversion_set.get(version="0.0.1")
+        self.assertTrue(
+            version.approved,
+            "Version uploaded by a trusted user (can_approve) should be approved automatically.",
+        )
+
     def tearDown(self):
         self.client.logout()

qgis-app/plugins/tests/test_plugin_upload.py

@@ -86,5 +86,91 @@ def test_plugin_upload_form(self):
             mail.outbox[0].from_email,
             settings.DEFAULT_FROM_EMAIL
         )
+
+    @patch("plugins.tasks.generate_plugins_xml", new=do_nothing)
+    @patch("plugins.validator._check_url_link", new=do_nothing)
+    def test_new_version_not_auto_approved_for_untrusted_user_on_approved_plugin(self):
+        """
+        Security: a new version must NOT be auto-approved because the plugin is
+        approved. Only user trust (can_approve permission) should grant approval.
+        """
+        self.client.login(username='testuser', password='testpassword')
+
+        # Upload the initial plugin
+        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"
+            )
+        self.client.post(self.url, {'package': uploaded_file})
+
+        plugin = Plugin.objects.get(name='Test Plugin')
+
+        # Simulate staff approval of the existing version
+        version = PluginVersion.objects.get(plugin=plugin, version='0.0.1')
+        version.approved = True
+        version.save()
+        self.assertTrue(version.approved)
+
+        # Upload a new version as the same untrusted user
+        valid_plugin_v2 = os.path.join(TESTFILE_DIR, "valid_plugin_0.0.2.zip_")
+        with open(valid_plugin_v2, "rb") as file:
+            uploaded_file_v2 = SimpleUploadedFile(
+                "valid_plugin_0.0.2.zip_", file.read(), content_type="application/zip"
+            )
+        response = self.client.post(self.url, {'package': uploaded_file_v2})
+
+        self.assertEqual(response.status_code, 302)
+        new_version = PluginVersion.objects.get(plugin=plugin, version='0.0.2')
+        self.assertFalse(
+            new_version.approved,
+            "New version must NOT be auto-approved just because the plugin is approved; "
+            "only user trust (can_approve) should grant approval.",
+        )
+
+    @patch("plugins.tasks.generate_plugins_xml", new=do_nothing)
+    @patch("plugins.validator._check_url_link", new=do_nothing)
+    def test_new_version_auto_approved_for_trusted_user(self):
+        """
+        A user with can_approve permission should have their uploaded version
+        approved automatically via the plugin_upload view.
+        """
+        from django.contrib.auth.models import Permission
+        from django.contrib.contenttypes.models import ContentType
+        from plugins.models import Plugin as PluginModel
+
+        ct = ContentType.objects.get_for_model(PluginModel)
+        perm = Permission.objects.get(codename="can_approve", content_type=ct)
+        self.user.user_permissions.add(perm)
+        # Refresh to bust Django's permission cache
+        self.user = self.user.__class__.objects.get(pk=self.user.pk)
+
+        self.client.login(username="testuser", password="testpassword")
+
+        # Upload the initial plugin
+        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"
+            )
+        self.client.post(self.url, {"package": uploaded_file})
+
+        plugin = Plugin.objects.get(name="Test Plugin")
+
+        # Upload a second version as the trusted user
+        valid_plugin_v2 = os.path.join(TESTFILE_DIR, "valid_plugin_0.0.2.zip_")
+        with open(valid_plugin_v2, "rb") as file:
+            uploaded_file_v2 = SimpleUploadedFile(
+                "valid_plugin_0.0.2.zip_", file.read(), content_type="application/zip"
+            )
+        response = self.client.post(self.url, {"package": uploaded_file_v2})
+
+        self.assertEqual(response.status_code, 302)
+        new_version = PluginVersion.objects.get(plugin=plugin, version="0.0.2")
+        self.assertTrue(
+            new_version.approved,
+            "Version uploaded by a trusted user (can_approve) should be approved automatically.",
+        )
+
     def tearDown(self):
         self.client.logout()

qgis-app/plugins/tests/test_token_auth.py

@@ -238,6 +238,95 @@ def test_update_version_with_invalid_token(self):
             ).exists()
         )
 
+    def test_new_version_not_auto_approved_for_untrusted_user_on_approved_plugin(self):
+        """
+        Security: uploading a new version via token must NOT be auto-approved
+        because the plugin already has an approved version. Only the user's
+        can_approve permission should grant approval.
+        """
+        # Approve the existing version to simulate an approved plugin
+        version = PluginVersion.objects.get(plugin__name="Test Plugin", version="0.0.1")
+        version.approved = True
+        version.save()
+
+        # Generate a token for the untrusted user
+        self.client.post(self.url_token_create, {})
+        outstanding_token = OutstandingToken.objects.last().token
+        refresh = RefreshToken(outstanding_token)
+        refresh["plugin_id"] = self.plugin.pk
+        refresh["refresh_jti"] = refresh["jti"]
+        access_token = str(refresh.access_token)
+
+        self.client.logout()
+
+        valid_plugin = os.path.join(TESTFILE_DIR, "valid_plugin_0.0.2.zip_")
+        with open(valid_plugin, "rb") as file:
+            uploaded_file = SimpleUploadedFile(
+                "valid_plugin_0.0.2.zip_", file.read(), content_type="application/zip_"
+            )
+
+        c = Client(HTTP_AUTHORIZATION=f"Bearer {access_token}")
+        response = c.post(self.url_add_version, {"package": uploaded_file})
+
+        self.assertEqual(response.status_code, 201)
+        data = response.json()
+        self.assertTrue(data.get("success"))
+
+        new_version = PluginVersion.objects.get(plugin__name="Test Plugin", version="0.0.2")
+        self.assertFalse(
+            new_version.approved,
+            "New version must NOT be auto-approved just because the plugin is approved; "
+            "only user trust (can_approve) should grant approval.",
+        )
+        self.assertFalse(data.get("approved"))
+        self.assertIn("approval_message", data)
+
+    def test_new_version_auto_approved_for_trusted_token_user(self):
+        """
+        A token whose owner holds can_approve should result in an approved version,
+        regardless of whether request.user is AnonymousUser (token path).
+        """
+        from django.contrib.auth.models import Permission
+        from django.contrib.contenttypes.models import ContentType
+        from plugins.models import Plugin as PluginModel
+
+        # Grant can_approve to the token user
+        ct = ContentType.objects.get_for_model(PluginModel)
+        perm = Permission.objects.get(codename="can_approve", content_type=ct)
+        self.user.user_permissions.add(perm)
+        # Refresh to bust the permission cache
+        self.user = self.user.__class__.objects.get(pk=self.user.pk)
+
+        # Generate a token for the now-trusted user
+        self.client.post(self.url_token_create, {})
+        outstanding_token = OutstandingToken.objects.last().token
+        refresh = RefreshToken(outstanding_token)
+        refresh["plugin_id"] = self.plugin.pk
+        refresh["refresh_jti"] = refresh["jti"]
+        access_token = str(refresh.access_token)
+
+        self.client.logout()
+
+        valid_plugin = os.path.join(TESTFILE_DIR, "valid_plugin_0.0.2.zip_")
+        with open(valid_plugin, "rb") as file:
+            uploaded_file = SimpleUploadedFile(
+                "valid_plugin_0.0.2.zip_", file.read(), content_type="application/zip_"
+            )
+
+        c = Client(HTTP_AUTHORIZATION=f"Bearer {access_token}")
+        response = c.post(self.url_add_version, {"package": uploaded_file})
+
+        self.assertEqual(response.status_code, 201)
+        data = response.json()
+        self.assertTrue(data.get("success"))
+
+        new_version = PluginVersion.objects.get(plugin__name="Test Plugin", version="0.0.2")
+        self.assertTrue(
+            new_version.approved,
+            "Version uploaded via token by a trusted user (can_approve) should be approved.",
+        )
+        self.assertTrue(data.get("approved"))
+
 
 class APIResponseTestCase(TestCase):
     """Test cases for API response improvements"""

qgis-app/plugins/views.py

@@ -557,8 +557,7 @@ def plugin_upload(request):
                     "version": form.cleaned_data.get("version"),
                     "created_by": request.user,
                     "package": form.cleaned_data.get("package"),
-                    "approved": request.user.has_perm("plugins.can_approve")
-                    or plugin.approved,
+                    "approved": request.user.has_perm("plugins.can_approve"),
                     "experimental": form.cleaned_data.get("experimental", False),
                     "changelog": form.cleaned_data.get("changelog", ""),
                     "external_deps": form.cleaned_data.get("external_deps", ""),
@@ -1563,7 +1562,14 @@ def _version_create(request, plugin, version):
     contained in the package metadata
     """
     is_api_request = getattr(version, "is_from_token", False)
-    is_trusted = request.user.has_perm("plugins.can_approve") or plugin.approved
+    if is_api_request:
+        # For token-based uploads, resolve the token owner and check their permissions.
+        # request.user is AnonymousUser in this path since authentication is via JWT token,
+        # not Django session.
+        token_user = request.plugin_token.token.user
+        is_trusted = token_user.has_perm("plugins.can_approve")
+    else:
+        is_trusted = request.user.has_perm("plugins.can_approve")
     if request.method == "POST":
 
         form = PluginVersionForm(