Merge branch 'dev' of https://github.com/MaibornWolff/SecObserve into stackable

Author: dervoeti

.github/workflows/scan_sca_current.yml

@@ -16,7 +16,7 @@ jobs:
         name: Checkout
         uses: actions/checkout@08c6903cd8c0fde910a37f88322edcfb5dd907a8 # v5.0.0
         with:
-          ref: 'v1.40.0'
+          ref: 'v1.41.0'
       -
         name: Run SCA vulnerability scanners
         uses: MaibornWolff/secobserve_actions_templates/actions/vulnerability_scanner@936a764a4e82cc89772941e082ba24c371c6ef90 # 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@4e94bd11f71e507f7f87df81788dff88d1dacbfb # v4.31.0
+        uses: github/codeql-action/upload-sarif@0499de31b99561a6d14a36a5f662c2a54f91beee # v4.31.2
         with:
           sarif_file: results.sarif

backend/application/__init__.py

@@ -1,4 +1,4 @@
-__version__ = "1.40.0"
+__version__ = "1.41.0"
 
 import pymysql
 

backend/application/access_control/api/filters.py

@@ -195,9 +195,9 @@ class ApiTokenFilter(FilterSet):
 
     ordering = OrderingFilter(
         # tuple-mapping retains order
-        fields=(("user__username", "name"),),
+        fields=(("user__username", "name"), ("user", "user")),
     )
 
     class Meta:
         model = API_Token
-        fields = ["name"]
+        fields = ["name", "user"]

backend/application/access_control/api/serializers.py

@@ -1,3 +1,4 @@
+import re
 from typing import Any, Optional
 
 from rest_framework.serializers import (
@@ -14,6 +15,7 @@
     Authorization_Group_Member,
     User,
 )
+from application.access_control.queries.api_token import get_api_tokens_for_user
 from application.access_control.queries.authorization_group_member import (
     get_authorization_group_member,
 )
@@ -111,6 +113,7 @@ class UserSerializer(UserListSerializer):
     has_authorization_groups = SerializerMethodField()
     has_product_group_members = SerializerMethodField()
     has_product_members = SerializerMethodField()
+    has_api_tokens = SerializerMethodField()
 
     class Meta:
         model = User
@@ -136,6 +139,7 @@ class Meta:
             "has_authorization_groups",
             "has_product_group_members",
             "has_product_members",
+            "has_api_tokens",
         ]
 
     def to_representation(self, instance: User) -> dict[str, Any]:
@@ -146,6 +150,7 @@ def to_representation(self, instance: User) -> dict[str, Any]:
             data.pop("has_authorization_groups")
             data.pop("has_product_group_members")
             data.pop("has_product_members")
+            data.pop("has_api_tokens")
 
         return data
 
@@ -161,6 +166,9 @@ def get_has_product_group_members(self, obj: User) -> bool:
     def get_has_product_members(self, obj: User) -> bool:
         return Product_Member.objects.filter(user=obj, product__is_product_group=False).exists()
 
+    def get_has_api_tokens(self, obj: User) -> bool:
+        return get_api_tokens_for_user(obj).exists()
+
 
 class UserUpdateSerializer(ModelSerializer):
     class Meta:
@@ -297,15 +305,17 @@ def get_name(self, obj: API_Token) -> str:
         return obj.user.username
 
     def get_product(self, obj: API_Token) -> Optional[int]:
-        product_member = Product_Member.objects.filter(user=obj.user, product__is_product_group=False).first()
-        if product_member:
-            return product_member.product.pk
+        if re.match("-product-\\d*-api_token-", obj.user.username):
+            product_member = Product_Member.objects.filter(user=obj.user, product__is_product_group=False).first()
+            if product_member:
+                return product_member.product.pk
         return None
 
     def get_product_group(self, obj: API_Token) -> Optional[int]:
-        product_member = Product_Member.objects.filter(user=obj.user, product__is_product_group=True).first()
-        if product_member:
-            return product_member.product.pk
+        if re.match("-product-\\d*-api_token-", obj.user.username):
+            product_member = Product_Member.objects.filter(user=obj.user, product__is_product_group=True).first()
+            if product_member:
+                return product_member.product.pk
         return None
 
 

backend/application/access_control/api/views.py

@@ -21,7 +21,7 @@
 from rest_framework.exceptions import PermissionDenied, ValidationError
 from rest_framework.filters import SearchFilter
 from rest_framework.mixins import ListModelMixin
-from rest_framework.permissions import IsAdminUser, IsAuthenticated
+from rest_framework.permissions import IsAuthenticated
 from rest_framework.request import Request
 from rest_framework.response import Response
 from rest_framework.serializers import BaseSerializer
@@ -60,6 +60,7 @@
     JWT_Secret,
     User,
 )
+from application.access_control.queries.api_token import get_api_tokens
 from application.access_control.queries.authorization_group import (
     get_authorization_groups,
 )
@@ -249,8 +250,11 @@ def get_queryset(self) -> QuerySet[Authorization_Group_Member]:
 class ApiTokenViewSet(ListModelMixin, GenericViewSet):
     serializer_class = ApiTokenSerializer
     filterset_class = ApiTokenFilter
-    permission_classes = (IsAuthenticated, IsAdminUser)
-    queryset = API_Token.objects.all()
+    permission_classes = (IsAuthenticated,)
+    queryset = API_Token.objects.none()
+
+    def get_queryset(self) -> QuerySet[API_Token]:
+        return get_api_tokens()
 
 
 class CreateUserAPITokenView(APIView):

backend/application/access_control/queries/api_token.py

@@ -0,0 +1,32 @@
+from django.db.models.query import QuerySet
+
+from application.access_control.models import API_Token, User
+from application.access_control.services.current_user import get_current_user
+
+
+def get_api_tokens() -> QuerySet[API_Token]:
+    user = get_current_user()
+
+    if user is None:
+        return API_Token.objects.none()
+
+    api_tokens = API_Token.objects.all()
+
+    if user.is_superuser:
+        return api_tokens
+
+    return api_tokens.filter(user=user)
+
+
+def get_api_tokens_for_user(given_user: User) -> QuerySet[API_Token]:
+    current_user = get_current_user()
+
+    if current_user is None:
+        return API_Token.objects.none()
+
+    api_tokens = API_Token.objects.filter(user=given_user)
+
+    if current_user.is_superuser:
+        return api_tokens
+
+    return api_tokens if current_user == given_user else API_Token.objects.none()

backend/application/core/signals.py

@@ -6,6 +6,7 @@
 from django.dispatch import receiver
 from huey.contrib.djhuey import db_task, lock_task
 
+from application.access_control.models import API_Token, User
 from application.access_control.services.current_user import get_current_user
 from application.authorization.services.roles_permissions import Roles
 from application.commons.models import Settings
@@ -29,6 +30,18 @@ def observation_pre_save(sender: Any, instance: Observation, **kwargs: Any) -> N
     set_product_flags(instance)
 
 
+@receiver(post_delete, sender=Product)
+def product_post_delete(sender: Any, instance: Product, **kwargs: Any) -> None:  # pylint: disable=unused-argument
+    # sender is needed according to Django documentation
+    try:
+        user = User.objects.get(username=f"-product-{instance.pk}-api_token-")
+        api_token = API_Token.objects.get(user=user)
+        api_token.delete()
+        user.delete()
+    except (User.DoesNotExist, API_Token.DoesNotExist):
+        pass
+
+
 @receiver(post_save, sender=Product)
 def product_post_save(
     sender: Any, instance: Product, created: bool, **kwargs: Any  # pylint: disable=unused-argument

backend/application/import_observations/parsers/sarif/parser.py

@@ -236,7 +236,7 @@ def get_description(  # pylint: disable=too-many-branches
     ) -> str:
         description = ""
 
-        sarif_message_text = result.get("message", {}).get("text")
+        sarif_message_text = result.get("message", {}).get("text", "")
         if sarif_message_text and not sarif_scanner.lower().startswith("trivy"):
             # Message text of Trivy has only redundant information
             description += f"{sarif_message_text}\n\n"