Merge branch 'bugfix' into os-messaging-banner

Author: Maffooch

dojo/api_v2/permissions.py

@@ -473,7 +473,13 @@ def has_permission(self, request, view):
             # Raise an explicit drf exception here
             raise ValidationError(e)
         if engagement := converted_dict.get("engagement"):
-            # existing engagement, nothing special to check
+            # Validate the resolved engagement's parent chain matches any provided identifiers
+            if (product := converted_dict.get("product")) and engagement.product_id != product.id:
+                msg = "The provided identifiers are inconsistent — the engagement does not belong to the specified product."
+                raise ValidationError(msg)
+            if (engagement_name := converted_dict.get("engagement_name")) and engagement.name != engagement_name:
+                msg = "The provided identifiers are inconsistent — the engagement name does not match the specified engagement."
+                raise ValidationError(msg)
             return user_has_permission(
                 request.user, engagement, Permissions.Import_Scan_Result,
             )
@@ -764,6 +770,11 @@ def has_permission(self, request, view):
         try:
             converted_dict = auto_create.convert_querydict_to_dict(request.data)
             auto_create.process_import_meta_data_from_dict(converted_dict)
+            # engagement is not a declared field on ReImportScanSerializer and will be
+            # stripped during validation — don't use it in the permission check either,
+            # so the permission check resolves targets the same way execution does
+            converted_dict.pop("engagement", None)
+            converted_dict.pop("engagement_id", None)
             # Get an existing product
             converted_dict["product_type"] = auto_create.get_target_product_type_if_exists(**converted_dict)
             converted_dict["product"] = auto_create.get_target_product_if_exists(**converted_dict)
@@ -774,7 +785,20 @@ def has_permission(self, request, view):
             raise ValidationError(e)
 
         if test := converted_dict.get("test"):
-            # existing test, nothing special to check
+            # Validate the resolved test's parent chain matches any provided identifiers
+            if (product := converted_dict.get("product")) and test.engagement.product_id != product.id:
+                msg = "The provided identifiers are inconsistent — the test does not belong to the specified product."
+                raise ValidationError(msg)
+            if (engagement := converted_dict.get("engagement")) and test.engagement_id != engagement.id:
+                msg = "The provided identifiers are inconsistent — the test does not belong to the specified engagement."
+                raise ValidationError(msg)
+            # Also validate by name when the objects were not resolved (e.g. names that match no existing record)
+            if not converted_dict.get("product") and (product_name := converted_dict.get("product_name")) and test.engagement.product.name != product_name:
+                msg = "The provided identifiers are inconsistent — the test does not belong to the specified product."
+                raise ValidationError(msg)
+            if not converted_dict.get("engagement") and (engagement_name := converted_dict.get("engagement_name")) and test.engagement.name != engagement_name:
+                msg = "The provided identifiers are inconsistent — the test does not belong to the specified engagement."
+                raise ValidationError(msg)
             return user_has_permission(
                 request.user, test, Permissions.Import_Scan_Result,
             )
@@ -1181,7 +1205,10 @@ def check_auto_create_permission(
         raise ValidationError(msg)
 
     if engagement:
-        # existing engagement, nothing special to check
+        # Validate the resolved engagement's parent chain matches any provided names
+        if product is not None and engagement.product_id != product.id:
+            msg = "The provided identifiers are inconsistent — the engagement does not belong to the specified product."
+            raise ValidationError(msg)
         return user_has_permission(
             user, engagement, Permissions.Import_Scan_Result,
         )

dojo/importers/auto_create_context.py

@@ -181,6 +181,9 @@ def get_target_engagement_if_exists(
         """
         if engagement := get_object_or_none(Engagement, pk=engagement_id):
             logger.debug("Using existing engagement by id: %s", engagement_id)
+            if product is not None and engagement.product_id != product.id:
+                msg = "The provided identifiers are inconsistent — the engagement does not belong to the specified product."
+                raise ValueError(msg)
             return engagement
         # if there's no product, then for sure there's no engagement either
         if product is None:
@@ -203,6 +206,9 @@ def get_target_test_if_exists(
         """
         if test := get_object_or_none(Test, pk=test_id):
             logger.debug("Using existing Test by id: %s", test_id)
+            if engagement is not None and test.engagement_id != engagement.id:
+                msg = "The provided identifiers are inconsistent — the test does not belong to the specified engagement."
+                raise ValueError(msg)
             return test
         # If the engagement is not supplied, we cannot do anything
         if not engagement:

dojo/templates/dojo/snippets/endpoints.html

@@ -72,13 +72,13 @@ <h6>Mitigated Endpoints / Systems ({{ finding.mitigated_endpoint_count }})</h6>
                                             {% if V3_FEATURE_LOCATIONS %}
                                                 <td style="word-break: break-word">{{ endpoint.location }}{% if endpoint.is_broken %} <span data-toggle="tooltip" title="Endpoint is broken. Check documentation to look for fix process" >&#128681;</span>{% endif %}</td>
                                                 <td>{{ endpoint.status }}</td>
-                                                <td>{{ endpoint.auditor|safe }}</td>
+                                                <td>{{ endpoint.auditor }}</td>
                                                 <td>{{ endpoint.audit_time|date }}</td>
                                             {% else %}
-                                                {% comment %} TODO: Delete this after the move to Locations {% endcomment %}    
+                                                {% comment %} TODO: Delete this after the move to Locations {% endcomment %}
                                                 <td style="word-break: break-word">{{ endpoint }}{% if endpoint.endpoint.is_broken %} <span data-toggle="tooltip" title="Endpoint is broken. Check documentation to look for fix process" >&#128681;</span>{% endif %}</td>
                                                 <td>{{ endpoint.status }}</td>
-                                                <td>{{ endpoint.mitigated_by|safe }}</td>
+                                                <td>{{ endpoint.mitigated_by }}</td>
                                                 <td>{{ endpoint.mitigated_time|date }}</td>
                                             {% endif %}
                                         </tr>
@@ -256,7 +256,7 @@ <h4>Mitigated Endpoints / Systems ({{ finding.mitigated_endpoint_count }})
                                                 {% include "dojo/snippets/tags.html" with tags=endpoint.location.tags.all %}
                                             </td>
                                             <td>{{ endpoint.get_status_display }}</td>
-                                            <td>{{ endpoint.auditor|safe }}</td>
+                                            <td>{{ endpoint.auditor }}</td>
                                             <td>{{ endpoint.audit_time|date }}</td>
                                         {% else %}
                                             {% comment %} TODO: Delete this after the move to Locations {% endcomment %}
@@ -265,7 +265,7 @@ <h4>Mitigated Endpoints / Systems ({{ finding.mitigated_endpoint_count }})
                                                 {% include "dojo/snippets/tags.html" with tags=endpoint.endpoint.tags.all %}
                                             </td>
                                             <td>{{ endpoint.status }}</td>
-                                            <td>{{ endpoint.mitigated_by|safe }}</td>
+                                            <td>{{ endpoint.mitigated_by }}</td>
                                             <td>{{ endpoint.mitigated_time|date }}</td>
                                         {% endif %}
                                     </tr>

dojo/tools/govulncheck/parser.py

@@ -40,6 +40,22 @@ def get_location(data, node):
     def get_version(data, node):
         return data["Requires"]["Modules"][str(node)]["Version"]
 
+    @staticmethod
+    def get_fix_info(affected_ranges):
+        for r in affected_ranges:
+            for event in r.get("events", []):
+                if "fixed" in event:
+                    return True, event["fixed"]
+        return False, ""
+
+    @staticmethod
+    def get_introduced_version(affected_ranges):
+        for r in affected_ranges:
+            for event in r.get("events", []):
+                if "introduced" in event:
+                    return event["introduced"]
+        return ""
+
     def get_finding_trace_info(self, data, osv_id):
         # Browse the findings to look for matching OSV-id. If the OSV-id is matching, extract traces.
         trace_info_strs = []
@@ -202,8 +218,12 @@ def get_findings(self, scan_file, test):
                         else:
                             title = f"{osv_data['id']} - {affected_package['name']}"
 
-                        affected_version = self.get_affected_version(data, osv_data["id"])
+                        fix_available, fix_version = self.get_fix_info(affected_ranges)
 
+                        affected_version = (
+                            self.get_affected_version(data, osv_data["id"])
+                            or self.get_introduced_version(affected_ranges)
+                        )
                         severity = elem["osv"].get("severity", SEVERITY)
 
                         d = {
@@ -215,6 +235,8 @@ def get_findings(self, scan_file, test):
                             "description": description,
                             "impact": impact,
                             "references": references,
+                            "fix_available": fix_available,
+                            "fix_version": fix_version,
                             "file_path": path,
                             "url": db_specific_url,
                             "unique_id_from_tool": osv_id,

unittests/test_rest_framework.py

@@ -3429,6 +3429,95 @@ def test_create_not_authorized_product_name_engagement_name_scan_type_title(self
             importer_mock.assert_not_called()
             reimporter_mock.assert_not_called()
 
+    # Security tests: verify that conflicting ID-based and name-based identifiers are rejected
+
+    @patch("dojo.importers.default_reimporter.DefaultReImporter.process_scan")
+    @patch("dojo.importers.default_importer.DefaultImporter.process_scan")
+    @patch("dojo.api_v2.permissions.user_has_permission")
+    def test_reimport_engagement_param_ignored_permission_checked_on_name_resolved_target(self, mock, importer_mock, reimporter_mock):
+        """
+        Engagement is not a declared field on ReImportScanSerializer — verify
+        the permission check uses the name-resolved target, not the engagement param.
+        """
+        mock.return_value = False
+        importer_mock.return_value = IMPORTER_MOCK_RETURN_VALUE
+        reimporter_mock.return_value = REIMPORTER_MOCK_RETURN_VALUE
+
+        with Path("tests/zap_sample.xml").open(encoding="utf-8") as testfile:
+            payload = {
+                "minimum_severity": "Low",
+                "active": True,
+                "verified": True,
+                "scan_type": "ZAP Scan",
+                "file": testfile,
+                # engagement=1 belongs to Product 2 Engagement 1, but it should be ignored
+                "engagement": 1,
+                # These names resolve to Product 2's Engagement 4 -> Test 4
+                "product_name": "Security How-to",
+                "engagement_name": "April monthly engagement",
+                "version": "1.0.0",
+            }
+            response = self.client.post(self.url, payload)
+            self.assertEqual(403, response.status_code, response.content[:1000])
+            # Permission must be checked on name-resolved Test 4 (in Engagement 4),
+            # NOT on Test 3 (which belongs to the engagement=1 param)
+            mock.assert_called_with(User.objects.get(username="admin"),
+                Test.objects.get(id=4),
+                Permissions.Import_Scan_Result)
+            importer_mock.assert_not_called()
+            reimporter_mock.assert_not_called()
+
+    @patch("dojo.importers.default_reimporter.DefaultReImporter.process_scan")
+    @patch("dojo.importers.default_importer.DefaultImporter.process_scan")
+    def test_reimport_with_test_id_mismatched_product_name_is_rejected(self, importer_mock, reimporter_mock):
+        """Sending test ID from one product with product_name from another must be rejected."""
+        importer_mock.return_value = IMPORTER_MOCK_RETURN_VALUE
+        reimporter_mock.return_value = REIMPORTER_MOCK_RETURN_VALUE
+
+        with Path("tests/zap_sample.xml").open(encoding="utf-8") as testfile:
+            payload = {
+                "minimum_severity": "Low",
+                "active": True,
+                "verified": True,
+                "scan_type": "ZAP Scan",
+                "file": testfile,
+                # Test 3 belongs to Engagement 1 -> Product 2 ("Security How-to")
+                "test": 3,
+                # But product_name points to Product 1 ("Python How-to")
+                "product_name": "Python How-to",
+                "version": "1.0.0",
+            }
+            response = self.client.post(self.url, payload)
+            self.assertEqual(400, response.status_code, response.content[:1000])
+            importer_mock.assert_not_called()
+            reimporter_mock.assert_not_called()
+
+    @patch("dojo.importers.default_reimporter.DefaultReImporter.process_scan")
+    @patch("dojo.importers.default_importer.DefaultImporter.process_scan")
+    def test_reimport_with_test_id_mismatched_engagement_name_is_rejected(self, importer_mock, reimporter_mock):
+        """Sending test ID from one engagement with engagement_name from another must be rejected."""
+        importer_mock.return_value = IMPORTER_MOCK_RETURN_VALUE
+        reimporter_mock.return_value = REIMPORTER_MOCK_RETURN_VALUE
+
+        with Path("tests/zap_sample.xml").open(encoding="utf-8") as testfile:
+            payload = {
+                "minimum_severity": "Low",
+                "active": True,
+                "verified": True,
+                "scan_type": "ZAP Scan",
+                "file": testfile,
+                # Test 3 belongs to Engagement 1 ("1st Quarter Engagement")
+                "test": 3,
+                # But engagement_name points to a different engagement
+                "product_name": "Security How-to",
+                "engagement_name": "April monthly engagement",
+                "version": "1.0.0",
+            }
+            response = self.client.post(self.url, payload)
+            self.assertEqual(400, response.status_code, response.content[:1000])
+            importer_mock.assert_not_called()
+            reimporter_mock.assert_not_called()
+
 
 @versioned_fixtures
 class ProductTypeTest(BaseClass.BaseClassTest):

unittests/tools/test_govulncheck_parser.py

@@ -127,6 +127,8 @@ def test_parse_new_version_many_findings_custom_severity(self):
                 self.assertIsNotNone(finding.impact)
                 self.assertIsNotNone(finding.description)
                 self.assertIsNotNone(finding.references)
+                self.assertTrue(finding.fix_available)
+                self.assertEqual("0.3.8", finding.fix_version)
 
     def test_parse_issue_14642(self):
         with (get_unit_tests_scans_path("govulncheck") / "issue_14642.json").open(encoding="utf-8") as testfile: