Merge remote-tracking branch 'origin/dev' into stackable

Author: dervoeti

.gitignore

@@ -28,3 +28,8 @@ __pycache__
 charts/secobserve/charts/postgresql-16.7.27.tgz
 charts/secobserve/my_values.yaml
 .kilo/
+
+openspec/
+.claude/
+.mcp.json
+CLAUDE.md

backend/application/access_control/api/filters.py

@@ -1,6 +1,6 @@
 from typing import Any, Optional
 
-from django.db.models import Exists, QuerySet
+from django.db.models import Exists, OuterRef, Q, QuerySet
 from django_filters import CharFilter, FilterSet, NumberFilter, OrderingFilter
 from rest_framework.request import Request
 
@@ -10,6 +10,23 @@
     Authorization_Group_Member,
     User,
 )
+from application.authorization.services.roles_permissions import Roles
+from application.core.models import (
+    Product,
+    Product_Authorization_Group_Member,
+    Product_Member,
+)
+
+
+def _get_product_and_group_ids(product_id: int) -> list[int]:
+    product = Product.objects.filter(id=product_id).select_related("product_group").first()
+    if product is None:
+        return []
+
+    product_ids = [product.id]
+    if product.product_group_id:
+        product_ids.append(product.product_group_id)
+    return product_ids
 
 
 class UserFilter(FilterSet):
@@ -23,6 +40,57 @@ class UserFilter(FilterSet):
     exclude_license_group = NumberFilter(field_name="exclude_license_group", method="get_exclude_license_group")
     exclude_license_policy = NumberFilter(field_name="exclude_license_policy", method="get_exclude_license_policy")
     exclude_product = NumberFilter(field_name="exclude_product", method="get_exclude_product")
+    member_of_product = NumberFilter(field_name="member_of_product", method="get_member_of_product")
+    assessment_approver_for_product = NumberFilter(
+        field_name="assessment_approver_for_product", method="get_assessment_approver_for_product"
+    )
+
+    def get_member_of_product(
+        self,
+        queryset: QuerySet,
+        name: Any,  # pylint: disable=unused-argument
+        value: Any,
+    ) -> QuerySet:
+        # Users who are members of the product - directly or via an authorization group - including
+        # membership inherited from the product's product group. The "authorization_groups" relation
+        # is traversed twice: User -> authorization groups -> products the group is assigned to.
+        if value is not None:
+            return queryset.filter(
+                Q(product_members__id=value)  # direct member of the product
+                | Q(product_members__products__id=value)  # direct member of the product group
+                | Q(authorization_groups__authorization_groups__id=value)  # via an authorization group of the product
+                | Q(authorization_groups__authorization_groups__products__id=value)  # via an auth group of the group
+            ).distinct()
+        return queryset
+
+    def get_assessment_approver_for_product(
+        self,
+        queryset: QuerySet,
+        name: Any,  # pylint: disable=unused-argument
+        value: Any,
+    ) -> QuerySet:
+        product_ids = _get_product_and_group_ids(value)
+        if not product_ids:
+            return queryset.none()
+
+        product_members = Product_Member.objects.filter(
+            product_id__in=product_ids,
+            user=OuterRef("pk"),
+            role__gte=Roles.Writer,
+        )
+        product_authorization_group_members = Product_Authorization_Group_Member.objects.filter(
+            product_id__in=product_ids,
+            authorization_group__users=OuterRef("pk"),
+            role__gte=Roles.Writer,
+        )
+        return (
+            queryset.annotate(
+                assessment_approver_product_member=Exists(product_members),
+                assessment_approver_group_member=Exists(product_authorization_group_members),
+            )
+            .filter(Q(assessment_approver_product_member=True) | Q(assessment_approver_group_member=True))
+            .distinct()
+        )
 
     def get_exclude_authorization_group(
         self,
@@ -116,6 +184,43 @@ class AuthorizationGroupFilter(FilterSet):
     exclude_license_group = NumberFilter(field_name="exclude_license_group", method="get_exclude_license_group")
     exclude_license_policy = NumberFilter(field_name="exclude_license_policy", method="get_exclude_license_policy")
     exclude_product = NumberFilter(field_name="exclude_product", method="get_exclude_product")
+    member_of_product = NumberFilter(field_name="member_of_product", method="get_member_of_product")
+    assessment_approver_for_product = NumberFilter(
+        field_name="assessment_approver_for_product", method="get_assessment_approver_for_product"
+    )
+
+    def get_member_of_product(
+        self,
+        queryset: QuerySet,
+        name: Any,  # pylint: disable=unused-argument
+        value: Any,
+    ) -> QuerySet:
+        # Authorization groups assigned to the product, including those inherited from its product group.
+        if value is not None:
+            return queryset.filter(
+                Q(authorization_groups__id=value)  # assigned to the product
+                | Q(authorization_groups__products__id=value)  # assigned to the product's product group
+            ).distinct()
+        return queryset
+
+    def get_assessment_approver_for_product(
+        self,
+        queryset: QuerySet,
+        name: Any,  # pylint: disable=unused-argument
+        value: Any,
+    ) -> QuerySet:
+        product_ids = _get_product_and_group_ids(value)
+        if not product_ids:
+            return queryset.none()
+
+        product_authorization_group_members = Product_Authorization_Group_Member.objects.filter(
+            product_id__in=product_ids,
+            authorization_group=OuterRef("pk"),
+            role__gte=Roles.Writer,
+        )
+        return queryset.annotate(
+            assessment_approver_product_authorization_group=Exists(product_authorization_group_members),
+        ).filter(assessment_approver_product_authorization_group=True)
 
     def get_exclude_license_group(
         self,

backend/application/authorization/services/authorization.py

@@ -22,6 +22,10 @@
     get_highest_role_of_product_authorization_group_members_for_user,
     get_product_member,
 )
+from application.core.services.assessment import (
+    assessment_approvers_configured,
+    is_user_designated_assessment_approver,
+)
 from application.import_observations.models import (
     Api_Configuration,
     Vulnerability_Check,
@@ -52,7 +56,11 @@ def user_has_permission(  # pylint: disable=too-many-return-statements,too-many-
         and permission not in Permissions.get_product_group_permissions()
     ):
         role = get_highest_user_role(obj, user)
-        return bool(role and role_has_permission(role, permission))
+        has_role_permission = bool(role and role_has_permission(role, permission))
+        # With designated approvers configured, only they (with an approval-capable role) may approve assessments.
+        if permission == Permissions.Observation_Log_Approval and assessment_approvers_configured(obj):
+            return has_role_permission and (role == Roles.Owner or is_user_designated_assessment_approver(obj, user))
+        return has_role_permission
 
     if isinstance(obj, Product) and obj.is_product_group:
         role = get_highest_user_role(obj, user)

backend/application/core/api/serializers_observation.py

@@ -655,6 +655,12 @@ class ObservationLogApprovalSerializer(Serializer):
     assessment_status = ChoiceField(choices=Assessment_Status.ASSESSMENT_STATUS_CHOICES_APPROVAL, required=True)
     rejection_remark = CharField(max_length=255, required=False, allow_blank=True)
     observation_log_comment = CharField(max_length=4096, required=False, allow_blank=True)
+    observation_log_vex_justification = ChoiceField(
+        choices=VEX_Justification.VEX_JUSTIFICATION_CHOICES,
+        required=False,
+        allow_blank=True,
+    )
+    observation_log_vex_remediations = JSONField(required=False, allow_null=True)
 
     def validate(self, attrs: dict) -> dict:
         if attrs.get("assessment_status") in [
@@ -671,8 +677,13 @@ def validate(self, attrs: dict) -> dict:
         if attrs.get("assessment_status") in [
             Assessment_Status.ASSESSMENT_STATUS_APPROVED,
             Assessment_Status.ASSESSMENT_STATUS_REJECTED,
-        ] and attrs.get("observation_log_comment"):
-            raise ValidationError("Comment for observation Log cannot be set with approval or rejection")
+        ]:
+            if attrs.get("observation_log_comment"):
+                raise ValidationError("Comment for observation Log cannot be set with approval or rejection")
+            if attrs.get("observation_log_vex_justification"):
+                raise ValidationError("VEX justification for observation log cannot be set with approval or rejection")
+            if attrs.get("observation_log_vex_remediations"):
+                raise ValidationError("VEX remediation for observation log cannot be set with approval or rejection")
 
         if attrs.get("assessment_status") == Assessment_Status.ASSESSMENT_STATUS_APPROVED_WITH_EDITS and not attrs.get(
             "observation_log_comment"