chore: merge for release 1.11.1

.github/workflows/scan_sca_current.yml

@@ -15,7 +15,7 @@ jobs:
         name: Checkout
         uses: actions/checkout@b4ffde65f46336ab88eb53be808477a3936bae11 # v4.1.1
         with:
-          ref: 'v1.11.0'
+          ref: 'v1.11.1'
       -
         name: Run SCA vulnerability scanners
         uses: MaibornWolff/secobserve_actions_templates/actions/vulnerability_scanner@cd1288ce6cb16c1b41bea98f60c275c0fc103166 # main

.github/workflows/scorecard.yml

@@ -67,6 +67,6 @@ jobs:
 
       # Upload the results to GitHub's code scanning dashboard.
       - name: "Upload to code-scanning"
-        uses: github/codeql-action/upload-sarif@4355270be187e1b672a7a1c7c7bae5afdc1ab94a # v3.24.10
+        uses: github/codeql-action/upload-sarif@df5a14dc28094dc936e103b37d749c6628682b60 # v3.25.0
         with:
           sarif_file: results.sarif

backend/application/__init__.py

@@ -1,4 +1,4 @@
-__version__ = "1.11.0"
+__version__ = "1.11.1"
 
 import pymysql
 

backend/application/access_control/services/product_api_token.py

@@ -14,13 +14,22 @@ def create_product_api_token(product: Product, role: Roles) -> str:
     product_user_name = _get_product_user_name(product)
     user = get_user_by_username(product_user_name)
     if user:
-        raise ValidationError("Only one API token per product is allowed.")
+        try:
+            user.api_token  # pylint: disable=pointless-statement
+            # This statement raises an exception if the user has no API token.
+            raise ValidationError("Only one API token per product is allowed.")
+        except API_Token.DoesNotExist:
+            pass
 
     api_token, api_token_hash = generate_api_token_hash()
 
-    user = User(username=product_user_name, is_active=True)
+    if user:
+        user.is_active = True
+    else:
+        user = User(username=product_user_name, is_active=True)
     user.set_unusable_password()
     user.save()
+
     Product_Member(product=product, user=user, role=role).save()
     API_Token(user=user, api_token_hash=api_token_hash).save()
 
@@ -33,15 +42,18 @@ def revoke_product_api_token(product: Product) -> None:
     if not user:
         return
 
-    api_tokens = API_Token.objects.filter(user=user)
-    for api_token in api_tokens:
+    try:
+        api_token = user.api_token
         api_token.delete()
+    except API_Token.DoesNotExist:
+        pass
 
     product_member = get_product_member(product, user)
     if product_member:
         product_member.delete()
 
-    user.delete()
+    user.is_active = False
+    user.save()
 
 
 @dataclass

backend/application/core/api/permissions.py

@@ -129,6 +129,9 @@ def has_object_permission(self, request, view, obj):
 
 class UserHasObservationPermission(BasePermission):
     def has_permission(self, request, view):
+        if request.path.endswith("/bulk_assessment/"):
+            return True
+
         return check_post_permission(
             request, Product, "product", Permissions.Observation_Create
         )

backend/application/core/api/views.py

@@ -487,6 +487,27 @@ def remove_assessment(self, request, pk=None):
 
         return Response()
 
+    @extend_schema(
+        methods=["POST"],
+        request=ObservationBulkAssessmentSerializer,
+        responses={HTTP_204_NO_CONTENT: None},
+    )
+    @action(detail=False, methods=["post"])
+    def bulk_assessment(self, request):
+        request_serializer = ObservationBulkAssessmentSerializer(data=request.data)
+        if not request_serializer.is_valid():
+            raise ValidationError(request_serializer.errors)
+
+        observations_bulk_assessment(
+            None,
+            request_serializer.validated_data.get("severity"),
+            request_serializer.validated_data.get("status"),
+            request_serializer.validated_data.get("comment"),
+            request_serializer.validated_data.get("observations"),
+            request_serializer.validated_data.get("vex_justification"),
+        )
+        return Response(status=HTTP_204_NO_CONTENT)
+
 
 class ObservationLogViewSet(GenericViewSet, ListModelMixin, RetrieveModelMixin):
     serializer_class = ObservationLogSerializer

backend/application/core/services/observations_bulk_actions.py

@@ -1,7 +1,11 @@
+from typing import Optional
+
 from django.db.models.query import QuerySet
 from django.utils import timezone
 from rest_framework.exceptions import ValidationError
 
+from application.access_control.services.authorization import user_has_permission
+from application.access_control.services.roles_permissions import Permissions
 from application.commons.services.global_request import get_current_user
 from application.core.models import Observation, Potential_Duplicate, Product
 from application.core.queries.observation import get_current_observation_log
@@ -15,7 +19,7 @@
 
 
 def observations_bulk_assessment(
-    product: Product,
+    product: Optional[Product],
     new_severity: str,
     new_status: str,
     comment: str,
@@ -88,17 +92,24 @@ def observations_bulk_mark_duplicates(
 
 
 def _check_observations(
-    product: Product, observation_ids: list[int]
+    product: Optional[Product], observation_ids: list[int]
 ) -> QuerySet[Observation]:
     observations = Observation.objects.filter(id__in=observation_ids)
     if len(observations) != len(observation_ids):
         raise ValidationError("Some observations do not exist")
 
     for observation in observations:
-        if observation.product != product:
-            raise ValidationError(
-                f"Observation {observation.pk} does not belong to product {product.pk}"
-            )
+        if product:
+            if observation.product != product:
+                raise ValidationError(
+                    f"Observation {observation.pk} does not belong to product {product.pk}"
+                )
+        else:
+            if not user_has_permission(observation, Permissions.Observation_Assessment):
+                raise ValidationError(
+                    f"First observation without assessment permission: {observation}"
+                )
+
         current_observation_log = get_current_observation_log(observation)
         if (
             current_observation_log

backend/pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "SecObserve"
-version = "1.11.0"
+version = "1.11.1"
 description = "SecObserve is an open source vulnerability management system for software development and cloud environments."
 license = "BSD-3-Clause"
 authors = [
@@ -78,7 +78,7 @@ types-PyMySQL = "1.1.0.1"  # https://github.com/python/typeshed
 django-extensions = "3.2.3"  # https://github.com/django-extensions/django-extensions
 
 [tool.poetry.group.prod.dependencies]
-gunicorn = "21.2.0"  # https://github.com/benoitc/gunicorn
+gunicorn = "22.0.0"  # https://github.com/benoitc/gunicorn
 
 [tool.poetry.group.unittests.dependencies]
 coverage = "7.4.4"  # https://github.com/nedbat/coveragepy

backend/unittests/access_control/services/test_product_api_token.py

@@ -17,18 +17,46 @@
 class TestProductApiToken(BaseTestCase):
     @patch("application.access_control.services.product_api_token.get_user_by_username")
     def test_create_product_api_token_exists(self, mock):
-        mock.return_value = User()
+        user = User(username="username", full_name="full_name")
+        api_token = API_Token(user=user, api_token_hash="hash")
+        mock.return_value = user
 
-        with self.assertRaises(ValidationError):
+        with self.assertRaises(ValidationError) as e:
             create_product_api_token(self.product_1, Roles.Upload)
             mock.assert_called_with("-product-None-api_token-")
+            self.assertEqual("Only one API token per product is allowed.", str(e))
 
     @patch("application.access_control.services.product_api_token.get_user_by_username")
     @patch("application.access_control.models.API_Token.save")
     @patch("application.access_control.models.User.save")
     @patch("application.core.models.Product_Member.save")
     @patch("application.access_control.models.User.set_unusable_password")
-    def test_create_product_api_token_new(
+    def test_create_product_api_token_with_user(
+        self,
+        set_unusable_password_mock,
+        product_member_save_mock,
+        user_save_mock,
+        api_token_save_mock,
+        user_mock,
+    ):
+        user_mock.return_value = User()
+
+        api_token = create_product_api_token(self.product_1, Roles.Upload)
+
+        self.assertEqual(42, len(api_token))
+
+        user_mock.assert_called_with("-product-None-api_token-")
+        api_token_save_mock.assert_called()
+        user_save_mock.assert_called()
+        product_member_save_mock.assert_called()
+        set_unusable_password_mock.assert_called()
+
+    @patch("application.access_control.services.product_api_token.get_user_by_username")
+    @patch("application.access_control.models.API_Token.save")
+    @patch("application.access_control.models.User.save")
+    @patch("application.core.models.Product_Member.save")
+    @patch("application.access_control.models.User.set_unusable_password")
+    def test_create_product_api_token_without_user(
         self,
         set_unusable_password_mock,
         product_member_save_mock,
@@ -58,38 +86,31 @@ def test_revoke_product_api_token_not_exists(self, filter_mock, user_mock):
         filter_mock.assert_not_called()
 
     @patch("application.access_control.services.product_api_token.get_user_by_username")
-    @patch("application.access_control.models.API_Token.objects.filter")
     @patch("application.access_control.models.API_Token.delete")
-    @patch("application.access_control.models.User.delete")
+    @patch("application.access_control.models.User.save")
     @patch("application.core.models.Product_Member.delete")
     @patch("application.access_control.services.product_api_token.get_product_member")
     def test_revoke_product_api_token(
         self,
         get_product_member_mock,
         product_member_delete_mock,
-        user_delete_mock,
+        user_save_mock,
         api_token_delete_mock,
-        filter_mock,
         user_mock,
     ):
-        user = User()
+        user = User(username="username", full_name="full_name")
+        api_token = API_Token(user=user, api_token_hash="hash")
         user_mock.return_value = user
 
-        none_qs = API_Token.objects.none()
-        api_token_1 = API_Token()
-        api_token_2 = API_Token()
-        qs = list(chain(none_qs, [api_token_1, api_token_2]))
-        filter_mock.return_value = qs
-
         get_product_member_mock.return_value = Product_Member()
 
         revoke_product_api_token(self.product_1)
 
         user_mock.assert_called_with("-product-None-api_token-")
-        filter_mock.assert_called_with(user=user)
-        self.assertEqual(2, api_token_delete_mock.call_count)
-        self.assertEqual(1, product_member_delete_mock.call_count)
-        user_delete_mock.assert_called()
+        api_token_delete_mock.assert_called()
+        get_product_member_mock.assert_called_with(self.product_1, user)
+        product_member_delete_mock.assert_called()
+        user_save_mock.assert_called()
 
     @patch("application.access_control.services.product_api_token.get_user_by_username")
     def test_get_product_api_tokens_no_user(self, user_mock):

backend/unittests/core/services/test_observations_bulk_actions.py

@@ -110,10 +110,31 @@ def test_check_observation_not_in_product(self):
             "[ErrorDetail(string='Observation 1 does not belong to product 2', code='invalid')]",
         )
 
-    def test_check_observation_success(self):
+    def test_check_observation_product_success(self):
         product_1 = Product.objects.get(pk=1)
 
         observations = _check_observations(product_1, [1])
 
         self.assertEqual(len(observations), 1)
         self.assertEqual(observations[0], Observation.objects.get(pk=1))
+
+    @patch("application.core.services.observations_bulk_actions.user_has_permission")
+    def test_check_observation_no_product_no_permission(self, mock_user_has_permission):
+        mock_user_has_permission.return_value = False
+
+        with self.assertRaises(ValidationError) as e:
+            _check_observations(None, [1])
+
+        self.assertEqual(
+            str(e.exception),
+            "[ErrorDetail(string='First observation without assessment permission: db_product_internal / db_observation_internal', code='invalid')]",
+        )
+
+    @patch("application.core.services.observations_bulk_actions.user_has_permission")
+    def test_check_observation_no_product_success(self, mock_user_has_permission):
+        mock_user_has_permission.return_value = True
+
+        observations = _check_observations(None, [1])
+
+        self.assertEqual(len(observations), 1)
+        self.assertEqual(observations[0], Observation.objects.get(pk=1))

backend/unittests/vex/api/files/csaf_given_vulnerability.json

@@ -18,7 +18,7 @@
             "generator": {
                 "engine": {
                     "name": "SecObserve",
-                    "version": "1.11.0"
+                    "version": "1.11.1"
                 }
             },
             "id": "CSAF_2024_0001_0001",

backend/unittests/vex/api/files/csaf_given_vulnerability_update.json

@@ -18,7 +18,7 @@
             "generator": {
                 "engine": {
                     "name": "SecObserve",
-                    "version": "1.11.0"
+                    "version": "1.11.1"
                 }
             },
             "id": "CSAF_2024_0001_0002",

backend/unittests/vex/api/files/csaf_product_branches.json

@@ -18,7 +18,7 @@
             "generator": {
                 "engine": {
                     "name": "SecObserve",
-                    "version": "1.11.0"
+                    "version": "1.11.1"
                 }
             },
             "id": "CSAF_2024_0001_0001",

backend/unittests/vex/api/files/csaf_product_given_branch.json

@@ -18,7 +18,7 @@
             "generator": {
                 "engine": {
                     "name": "SecObserve",
-                    "version": "1.11.0"
+                    "version": "1.11.1"
                 }
             },
             "id": "CSAF_2024_0001_0001",

backend/unittests/vex/api/files/csaf_product_no_branch.json

@@ -18,7 +18,7 @@
             "generator": {
                 "engine": {
                     "name": "SecObserve",
-                    "version": "1.11.0"
+                    "version": "1.11.1"
                 }
             },
             "id": "CSAF_2024_0001_0001",

backend/unittests/vex/api/files/csaf_product_no_branch_update.json

@@ -18,7 +18,7 @@
             "generator": {
                 "engine": {
                     "name": "SecObserve",
-                    "version": "1.11.0"
+                    "version": "1.11.1"
                 }
             },
             "id": "CSAF_2024_0001_0002",