From f35c2d7979c42ca3fe9dd8de830f17147a336949 Mon Sep 17 00:00:00 2001 From: Sam Vader Date: Fri, 13 Nov 2026 11:11:12 -0600 Subject: [PATCH 1/2] fix(import): resolve the import target the same way in the permission check and the import The scan import permission checks read the raw request, while the import itself runs on the serializer's validated data. Two differences let them pick different targets: the serializer trims whitespace from the name fields and the raw read did not, and the import permission check read a response-only engagement field that the serializer drops. Trim the names when reading the raw request, and ignore that field in the check, so the check authorizes the same product and engagement the import writes to. --- dojo/authorization/api_permissions.py | 3 + dojo/importers/auto_create_context.py | 5 + .../test_apiv2_import_target_authorization.py | 115 ++++++++++++++++++ 3 files changed, 123 insertions(+) create mode 100644 unittests/test_apiv2_import_target_authorization.py diff --git a/dojo/authorization/api_permissions.py b/dojo/authorization/api_permissions.py index f42fcf7a4e1..7166e988aa4 100644 --- a/dojo/authorization/api_permissions.py +++ b/dojo/authorization/api_permissions.py @@ -554,6 +554,9 @@ def has_permission(self, request, view): # in this case and wrap them in a DRF exception try: converted_dict = auto_create.convert_querydict_to_dict(request.data) + # engagement_id is read-only on ImportScanSerializer and dropped during validation, + # so only the engagement field may name the target here, as it does at execution + converted_dict.pop("engagement_id", None) auto_create.process_import_meta_data_from_dict(converted_dict) # Get an existing product converted_dict["product_type"] = auto_create.get_target_product_type_if_exists(**converted_dict) diff --git a/dojo/importers/auto_create_context.py b/dojo/importers/auto_create_context.py index 3f2edc185e0..0f1dd656be6 100644 --- a/dojo/importers/auto_create_context.py +++ b/dojo/importers/auto_create_context.py @@ -412,5 +412,10 @@ def convert_querydict_to_dict( copy[key] = value if len(value) > 1 else value[0] else: copy[key] = value + # Trim names the way the serializers' CharFields do, so a permission check + # resolves the same objects the import later uses + for key in ("product_type_name", "product_name", "engagement_name", "test_title"): + if isinstance(copy.get(key), str): + copy[key] = copy[key].strip() # Convert to a regular dict return copy diff --git a/unittests/test_apiv2_import_target_authorization.py b/unittests/test_apiv2_import_target_authorization.py new file mode 100644 index 00000000000..c891e0e2c64 --- /dev/null +++ b/unittests/test_apiv2_import_target_authorization.py @@ -0,0 +1,115 @@ +""" +Regression tests: the scan-import permission checks must authorize the product the import writes to. + +The permission classes read the raw request, while the import runs on the serializer's +validated data. Both have to resolve the same target product and engagement. +""" +import datetime +import json + +from django.core.files.uploadedfile import SimpleUploadedFile +from django.utils import timezone +from rest_framework.authtoken.models import Token +from rest_framework.test import APIClient + +from dojo.models import Dojo_User, Engagement, Finding, Product, Product_Type, Test, Test_Type + +from .dojo_test_case import DojoTestCase + +SCAN_TYPE = "Generic Findings Import" + + +class TestImportTargetAuthorization(DojoTestCase): + + @classmethod + def setUpTestData(cls): + today = datetime.date(2026, 11, 13) + cls.member = Dojo_User.objects.create_user(username="import_target_member", is_active=True) + cls.token = Token.objects.create(user=cls.member) + + cls.own_product = Product.objects.create( + name="Import Target Own Product", + description="own", + prod_type=Product_Type.objects.create(name="Import Target Own Type"), + ) + cls.own_product.authorized_users.add(cls.member) + cls.own_engagement = Engagement.objects.create( + name="Import Target Own Engagement", product=cls.own_product, target_start=today, target_end=today, + ) + + cls.other_product = Product.objects.create( + name="Import Target Other Product", + description="other", + prod_type=Product_Type.objects.create(name="Import Target Other Type"), + ) + other_engagement = Engagement.objects.create( + name="Import Target Other Engagement", product=cls.other_product, target_start=today, target_end=today, + ) + other_test = Test.objects.create( + engagement=other_engagement, + test_type=Test_Type.objects.get_or_create(name=SCAN_TYPE)[0], + target_start=timezone.now(), + target_end=timezone.now(), + ) + cls.other_finding = Finding.objects.create( + test=other_test, title="Import Target Other Finding", severity="High", description="other", + reporter=cls.member, active=True, + ) + + def _post(self, path, file, **fields): + client = APIClient() + client.credentials(HTTP_AUTHORIZATION=f"Token {self.token.key}") + return client.post(f"/api/v2/{path}/", {"file": file, **fields}, format="multipart") + + def _import(self, findings, **fields): + report = json.dumps({"findings": findings}).encode() + return self._post( + "import-scan", + SimpleUploadedFile("report.json", report, "application/json"), + scan_type=SCAN_TYPE, + auto_create_context=True, + **fields, + ) + + def _import_into_other_product(self, findings, **fields): + return self._import( + findings, + engagement_id=self.own_engagement.id, + engagement_name=self.own_engagement.name, + product_name=f" {self.other_product.name}", + **fields, + ) + + def test_own_engagement_id_does_not_authorize_an_import_into_another_product(self): + response = self._import_into_other_product( + [{"title": "planted", "severity": "High", "description": "x", "date": "2026-11-13"}], + ) + self.assertEqual(response.status_code, 403, response.content[:500]) + self.assertFalse(Engagement.objects.filter(product=self.other_product, name=self.own_engagement.name).exists()) + + def test_close_old_findings_cannot_reach_another_product(self): + response = self._import_into_other_product( + [], close_old_findings=True, close_old_findings_product_scope=True, + ) + self.assertEqual(response.status_code, 403, response.content[:500]) + self.other_finding.refresh_from_db() + self.assertTrue(self.other_finding.active) + self.assertFalse(self.other_finding.is_mitigated) + + def test_endpoint_meta_import_checks_the_product_it_writes_to(self): + response = self._post( + "endpoint_meta_import", + SimpleUploadedFile("meta.csv", b"hostname,team\nimport-target.example.com,own\n", "text/csv"), + product_name=f" {self.other_product.name}", + product=self.own_product.id, + ) + self.assertEqual(response.status_code, 403, response.content[:500]) + + def test_padded_name_of_own_product_imports_into_own_engagement(self): + response = self._import( + [{"title": "own", "severity": "Low", "description": "x", "date": "2026-11-13"}], + engagement_name=self.own_engagement.name, + product_name=f" {self.own_product.name}", + ) + self.assertEqual(response.status_code, 201, response.content[:500]) + self.assertEqual(response.json()["engagement_id"], self.own_engagement.id) From b0230779944e023458569003a127f713bd1f331f Mon Sep 17 00:00:00 2001 From: Sam Vader Date: Fri, 13 Nov 2026 14:25:06 -0600 Subject: [PATCH 2/2] fix(import): ignore response-only ids in the reimport and meta import checks ReImportScanSerializer drops test_id and EndpointMetaImporterSerializer drops product_id, because both are response-only fields. The permission checks still read them from the raw request. Ignore them in the checks too, so each check resolves its target the same way as its serializer. Adds reimport tests, including one for a name with surrounding spaces, and a meta import test. --- dojo/authorization/api_permissions.py | 4 ++ .../test_apiv2_import_target_authorization.py | 45 +++++++++++++++++-- 2 files changed, 45 insertions(+), 4 deletions(-) diff --git a/dojo/authorization/api_permissions.py b/dojo/authorization/api_permissions.py index 7166e988aa4..8b2a1034416 100644 --- a/dojo/authorization/api_permissions.py +++ b/dojo/authorization/api_permissions.py @@ -630,6 +630,8 @@ def has_permission(self, request, view): # in this case and wrap them in a DRF exception try: converted_dict = auto_create.convert_querydict_to_dict(request.data) + # product_id is read-only on EndpointMetaImporterSerializer, so only product may name the target + converted_dict.pop("product_id", None) auto_create.process_import_meta_data_from_dict(converted_dict) # Get an existing product product = auto_create.get_target_product_if_exists(**converted_dict) @@ -758,6 +760,8 @@ def has_permission(self, request, view): # in this case and wrap them in a DRF exception try: converted_dict = auto_create.convert_querydict_to_dict(request.data) + # test_id is read-only on ReImportScanSerializer, so only test may name the target + converted_dict.pop("test_id", None) 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, diff --git a/unittests/test_apiv2_import_target_authorization.py b/unittests/test_apiv2_import_target_authorization.py index c891e0e2c64..4d9cc87ab6e 100644 --- a/unittests/test_apiv2_import_target_authorization.py +++ b/unittests/test_apiv2_import_target_authorization.py @@ -12,7 +12,7 @@ from rest_framework.authtoken.models import Token from rest_framework.test import APIClient -from dojo.models import Dojo_User, Engagement, Finding, Product, Product_Type, Test, Test_Type +from dojo.models import Development_Environment, Dojo_User, Engagement, Finding, Product, Product_Type, Test, Test_Type from .dojo_test_case import DojoTestCase @@ -36,6 +36,15 @@ def setUpTestData(cls): cls.own_engagement = Engagement.objects.create( name="Import Target Own Engagement", product=cls.own_product, target_start=today, target_end=today, ) + Development_Environment.objects.get_or_create(name="Development") + test_type = Test_Type.objects.get_or_create(name=SCAN_TYPE)[0] + cls.own_test = Test.objects.create( + engagement=cls.own_engagement, + test_type=test_type, + scan_type=SCAN_TYPE, + target_start=timezone.now(), + target_end=timezone.now(), + ) cls.other_product = Product.objects.create( name="Import Target Other Product", @@ -45,14 +54,14 @@ def setUpTestData(cls): other_engagement = Engagement.objects.create( name="Import Target Other Engagement", product=cls.other_product, target_start=today, target_end=today, ) - other_test = Test.objects.create( + cls.other_test = Test.objects.create( engagement=other_engagement, - test_type=Test_Type.objects.get_or_create(name=SCAN_TYPE)[0], + test_type=test_type, target_start=timezone.now(), target_end=timezone.now(), ) cls.other_finding = Finding.objects.create( - test=other_test, title="Import Target Other Finding", severity="High", description="other", + test=cls.other_test, title="Import Target Other Finding", severity="High", description="other", reporter=cls.member, active=True, ) @@ -71,6 +80,16 @@ def _import(self, findings, **fields): **fields, ) + def _reimport(self, **fields): + report = json.dumps({"findings": [{"title": "again", "severity": "Low", "description": "x", "date": "2026-11-13"}]}).encode() + return self._post( + "reimport-scan", + SimpleUploadedFile("report.json", report, "application/json"), + scan_type=SCAN_TYPE, + engagement_name=self.own_engagement.name, + **fields, + ) + def _import_into_other_product(self, findings, **fields): return self._import( findings, @@ -113,3 +132,21 @@ def test_padded_name_of_own_product_imports_into_own_engagement(self): ) self.assertEqual(response.status_code, 201, response.content[:500]) self.assertEqual(response.json()["engagement_id"], self.own_engagement.id) + + def test_padded_name_of_own_product_reimports_into_own_test(self): + response = self._reimport(product_name=f" {self.own_product.name}") + self.assertEqual(response.status_code, 201, response.content[:500]) + self.assertEqual(response.json()["test_id"], self.own_test.id) + + def test_reimport_ignores_test_id(self): + response = self._reimport(product_name=self.own_product.name, test_id=self.other_test.id) + self.assertEqual(response.status_code, 201, response.content[:500]) + self.assertEqual(response.json()["test_id"], self.own_test.id) + + def test_endpoint_meta_import_ignores_product_id(self): + response = self._post( + "endpoint_meta_import", + SimpleUploadedFile("meta.csv", b"hostname,team\nimport-target.example.com,own\n", "text/csv"), + product_id=self.own_product.id, + ) + self.assertEqual(response.status_code, 400, response.content[:500])