diff --git a/dojo/authorization/api_permissions.py b/dojo/authorization/api_permissions.py index f42fcf7a4e1..8b2a1034416 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) @@ -627,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) @@ -755,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/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..4d9cc87ab6e --- /dev/null +++ b/unittests/test_apiv2_import_target_authorization.py @@ -0,0 +1,152 @@ +""" +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 Development_Environment, 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, + ) + 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", + 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, + ) + cls.other_test = Test.objects.create( + engagement=other_engagement, + test_type=test_type, + target_start=timezone.now(), + target_end=timezone.now(), + ) + cls.other_finding = Finding.objects.create( + test=cls.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 _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, + 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) + + 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])