{% endif %}
+
+
{% include "dojo/snippets/endpoints.html" with finding=finding destination="UI" %}
diff --git a/dojo/tool_product/views.py b/dojo/tool_product/views.py
index def26f088d2..98f56d2f5ce 100644
--- a/dojo/tool_product/views.py
+++ b/dojo/tool_product/views.py
@@ -2,7 +2,7 @@
import logging
from django.contrib import messages
-from django.core.exceptions import BadRequest
+from django.core.exceptions import PermissionDenied
from django.http import HttpResponseRedirect
from django.shortcuts import get_object_or_404, render
from django.urls import reverse
@@ -60,10 +60,9 @@ def all_tool_product(request, pid):
@user_is_authorized(Product, Permissions.Product_Edit, "pid")
def edit_tool_product(request, pid, ttid):
product = get_object_or_404(Product, id=pid)
- tool_product = Tool_Product_Settings.objects.get(pk=ttid)
+ tool_product = get_object_or_404(Tool_Product_Settings, pk=ttid)
if tool_product.product != product:
- msg = f"Product {pid} does not fit to product of Tool_Product {tool_product.product.id}"
- raise BadRequest(msg)
+ raise PermissionDenied
if request.method == "POST":
tform = ToolProductSettingsForm(request.POST, instance=tool_product)
@@ -87,11 +86,10 @@ def edit_tool_product(request, pid, ttid):
@user_is_authorized(Product, Permissions.Product_Edit, "pid")
def delete_tool_product(request, pid, ttid):
- tool_product = Tool_Product_Settings.objects.get(pk=ttid)
+ tool_product = get_object_or_404(Tool_Product_Settings, pk=ttid)
product = get_object_or_404(Product, id=pid)
if tool_product.product != product:
- msg = f"Product {pid} does not fit to product of Tool_Product {tool_product.product.id}"
- raise BadRequest(msg)
+ raise PermissionDenied
if request.method == "POST":
DeleteToolProductSettingsForm(request.POST, instance=tool_product)
diff --git a/dojo/tools/blackduck/importer.py b/dojo/tools/blackduck/importer.py
index 7273770c6ec..ad93cce6d01 100644
--- a/dojo/tools/blackduck/importer.py
+++ b/dojo/tools/blackduck/importer.py
@@ -7,6 +7,8 @@
from collections.abc import Iterable
from pathlib import Path
+from dojo.tools.utils import safe_open_zip
+
from .model import BlackduckFinding
@@ -48,7 +50,7 @@ def _process_zipfile(self, report):
files = {}
security_issues = {}
- with zipfile.ZipFile(report) as zipf:
+ with safe_open_zip(report) as zipf:
for full_file_name in zipf.namelist():
file_name = full_file_name.split("/")[-1]
# Backwards compatibility, newer versions of Blackduck have a source file rather
diff --git a/dojo/tools/blackduck_component_risk/importer.py b/dojo/tools/blackduck_component_risk/importer.py
index 56f04f73eb0..e35089b5f13 100644
--- a/dojo/tools/blackduck_component_risk/importer.py
+++ b/dojo/tools/blackduck_component_risk/importer.py
@@ -4,6 +4,8 @@
import zipfile
from pathlib import Path
+from dojo.tools.utils import safe_open_zip
+
logger = logging.getLogger(__name__)
@@ -42,7 +44,7 @@ def _process_zipfile(self, report: Path) -> (dict, dict, dict):
components = {}
source = {}
try:
- with zipfile.ZipFile(report) as zipf:
+ with safe_open_zip(report) as zipf:
c_file = False
s_file = False
for full_file_name in zipf.namelist():
diff --git a/dojo/tools/fortify/fpr_parser.py b/dojo/tools/fortify/fpr_parser.py
index f406782bfb0..b13689e565e 100644
--- a/dojo/tools/fortify/fpr_parser.py
+++ b/dojo/tools/fortify/fpr_parser.py
@@ -1,12 +1,12 @@
import logging
import re
-import zipfile
from xml.etree.ElementTree import Element
from defusedxml import ElementTree
from dojo.models import Finding, Test
from dojo.tools.fortify.fortify_data import DescriptionData, RuleData, SnippetData, VulnerabilityData
+from dojo.tools.utils import safe_read_all_zip
logger = logging.getLogger(__name__)
@@ -26,13 +26,9 @@ def __init__(self):
pass
def parse_fpr(self, filename, test):
- if str(filename.__class__) == "":
- input_zip = zipfile.ZipFile(filename.name, "r")
- else:
- input_zip = zipfile.ZipFile(filename, "r")
# Read each file from the zip artifact into a dict with the format of
# filename: file_content
- zip_data = {name: input_zip.read(name) for name in input_zip.namelist()}
+ zip_data = safe_read_all_zip(filename)
root, self.namespaces = self.identify_root(zip_data, "audit.fvdl", "No audit.fvdl file found in the zip")
audit_log, self.namespaces_audit_log = self.identify_root(zip_data, "audit.xml")
return self.convert_vulnerabilities_to_findings(root, audit_log, test)
diff --git a/dojo/tools/ms_defender/parser.py b/dojo/tools/ms_defender/parser.py
index 49836ecc850..6739987f856 100644
--- a/dojo/tools/ms_defender/parser.py
+++ b/dojo/tools/ms_defender/parser.py
@@ -1,11 +1,11 @@
import json
import logging
-import zipfile
from django.conf import settings
from dojo.models import Endpoint, Finding
from dojo.tools.locations import LocationData
+from dojo.tools.utils import safe_read_all_zip
logger = logging.getLogger(__name__)
@@ -37,12 +37,7 @@ def get_findings(self, file, test):
logger.warning("Error parsing JSON file %s: %s", file.name, e)
return []
elif str(file.name).endswith(".zip"):
- if str(file.__class__) == "":
- input_zip = zipfile.ZipFile(file.name, "r")
- else:
- input_zip = zipfile.ZipFile(file, "r")
-
- zipdata = {name: input_zip.read(name) for name in input_zip.namelist()}
+ zipdata = safe_read_all_zip(file)
vulnerabilityfiles = []
machinefiles = []
for content in list(zipdata):
diff --git a/dojo/tools/sonarqube/parser.py b/dojo/tools/sonarqube/parser.py
index d9d4ecb6e55..6dd2c6285a4 100644
--- a/dojo/tools/sonarqube/parser.py
+++ b/dojo/tools/sonarqube/parser.py
@@ -1,6 +1,5 @@
import json
import logging
-import zipfile
from lxml import etree
@@ -8,6 +7,7 @@
from dojo.tools.sonarqube.sonarqube_restapi_zip import SonarQubeRESTAPIZIP
from dojo.tools.sonarqube.soprasteria_html import SonarQubeSoprasteriaHTML
from dojo.tools.sonarqube.soprasteria_json import SonarQubeSoprasteriaJSON
+from dojo.tools.utils import safe_read_all_zip
logger = logging.getLogger(__name__)
@@ -38,11 +38,7 @@ def get_findings(self, file, test):
return SonarQubeRESTAPIJSON().get_json_items(json_content, test, self.mode)
return []
if file.name.endswith(".zip"):
- if str(file.__class__) == "":
- input_zip = zipfile.ZipFile(file.name, "r")
- else:
- input_zip = zipfile.ZipFile(file, "r")
- zipdata = {name: input_zip.read(name) for name in input_zip.namelist()}
+ zipdata = safe_read_all_zip(file)
return SonarQubeRESTAPIZIP().get_items(zipdata, test, self.mode)
parser = etree.HTMLParser()
tree = etree.parse(file, parser)
diff --git a/dojo/tools/trivy_operator/checks_handler.py b/dojo/tools/trivy_operator/checks_handler.py
index 86b3952c1ca..506058ef6e9 100644
--- a/dojo/tools/trivy_operator/checks_handler.py
+++ b/dojo/tools/trivy_operator/checks_handler.py
@@ -23,9 +23,9 @@ def handle_checks(self, labels, checks, test):
for check in checks:
check_title = check.get("title")
check_severity = TRIVY_SEVERITIES[check.get("severity")]
- check_id = check.get("checkID", "0")
+ check_id = check.get("checkID") or "0"
check_references = ""
- if check_id != 0:
+ if check_id != "0":
check_references = (
"https://avd.aquasec.com/misconfig/kubernetes/"
+ check_id.lower()
@@ -60,7 +60,7 @@ def handle_checks(self, labels, checks, test):
)
finding_tags = [resource_namespace, check_category]
finding.unsaved_tags = [tag for tag in finding_tags if tag]
- if check_id:
+ if check_id != "0":
finding.unsaved_vulnerability_ids = [UniformTrivyVulnID().return_uniformed_vulnid(check_id)]
findings.append(finding)
return findings
diff --git a/dojo/tools/trivy_operator/compliance_handler.py b/dojo/tools/trivy_operator/compliance_handler.py
index a73c60c4dcd..d529e2b0f57 100644
--- a/dojo/tools/trivy_operator/compliance_handler.py
+++ b/dojo/tools/trivy_operator/compliance_handler.py
@@ -31,10 +31,7 @@ def handle_compliance(self, benchmarkreport, test):
check_severity = check.get("severity", "")
check_target = check.get("target", "")
check_title = check.get("title", "")
- if not check_severity:
- severity = TRIVY_SEVERITIES[check_severity]
- else:
- severity = TRIVY_SEVERITIES[result_severity]
+ severity = TRIVY_SEVERITIES[check_severity] if check_severity else TRIVY_SEVERITIES[result_severity]
description += "**result description:** " + result_description + "\n"
description += "**result id:** " + result_id + "\n"
description += "**result name:** " + result_name + "\n"
diff --git a/dojo/tools/utils.py b/dojo/tools/utils.py
index 23049469cec..2b462234487 100644
--- a/dojo/tools/utils.py
+++ b/dojo/tools/utils.py
@@ -1,8 +1,83 @@
+import io
import json
import logging
+import zipfile
logger = logging.getLogger(__name__)
+# Zip bomb protection limits
+MAX_ZIP_MEMBERS = 1000
+MAX_ZIP_MEMBER_SIZE = 512 * 1024 * 1024 # 512 MB per member (uncompressed)
+MAX_ZIP_TOTAL_SIZE = 1 * 1024 * 1024 * 1024 # 1 GB total (uncompressed)
+MAX_ZIP_RATIO = 100 # max compression ratio (uncompressed / compressed)
+
+
+def safe_open_zip(file):
+ """
+ Open a zip file with protection against zip bomb attacks.
+
+ Validates member count, per-member uncompressed size, total uncompressed
+ size, and compression ratios using the central-directory metadata before
+ any data is extracted.
+
+ Accepts a file-like object or an io.TextIOWrapper (in which case
+ file.name is used as the path).
+
+ Returns an open ZipFile. Use as a context manager or call .close()
+ explicitly when done.
+
+ Raises ValueError if any limit is exceeded.
+ """
+ zf = zipfile.ZipFile(file.name, "r") if isinstance(file, io.TextIOWrapper) else zipfile.ZipFile(file, "r")
+
+ infos = zf.infolist()
+
+ if len(infos) > MAX_ZIP_MEMBERS:
+ zf.close()
+ msg = f"Zip file contains {len(infos)} members, exceeding the limit of {MAX_ZIP_MEMBERS}."
+ raise ValueError(msg)
+
+ total_size = 0
+ for info in infos:
+ if info.file_size > MAX_ZIP_MEMBER_SIZE:
+ zf.close()
+ msg = (
+ f"Zip member '{info.filename}' has uncompressed size {info.file_size} bytes, "
+ f"exceeding the per-member limit of {MAX_ZIP_MEMBER_SIZE} bytes."
+ )
+ raise ValueError(msg)
+ if info.compress_size > 0 and (info.file_size / info.compress_size) > MAX_ZIP_RATIO:
+ zf.close()
+ ratio = info.file_size / info.compress_size
+ msg = (
+ f"Zip member '{info.filename}' has a compression ratio of "
+ f"{ratio:.1f}:1, exceeding the limit of {MAX_ZIP_RATIO}:1."
+ )
+ raise ValueError(msg)
+ total_size += info.file_size
+ if total_size > MAX_ZIP_TOTAL_SIZE:
+ zf.close()
+ msg = f"Zip file total uncompressed size exceeds the limit of {MAX_ZIP_TOTAL_SIZE} bytes."
+ raise ValueError(msg)
+
+ return zf
+
+
+def safe_read_all_zip(file):
+ """
+ Open a zip file safely and read all members into a dict {name: bytes}.
+
+ Applies the same zip bomb protections as safe_open_zip before reading
+ any data.
+
+ Raises ValueError if any limit is exceeded.
+ """
+ zf = safe_open_zip(file)
+ try:
+ return {name: zf.read(name) for name in zf.namelist()}
+ finally:
+ zf.close()
+
def get_npm_cwe(item_node):
"""
diff --git a/dojo/url/ui/views.py b/dojo/url/ui/views.py
index a879867ea75..fcf2226522f 100644
--- a/dojo/url/ui/views.py
+++ b/dojo/url/ui/views.py
@@ -559,26 +559,14 @@ def finding_location_bulk_update(request, finding_id):
if request.method == "POST":
# Get the list of endpoint IDs to update and the statuses to enable
finding_locations_to_update = request.POST.getlist("endpoints_to_update")
- status_list = FindingLocationStatus.values
- enable = [item for item in status_list if item in list(request.POST.keys())]
+ # Get the status
+ status = request.POST.get("bulk_status")
# Check that endpoints and statuses are selected before proceeding
- if finding_locations_to_update and len(enable) > 0:
+ if finding_locations_to_update and status in FindingLocationStatus:
# Iterate over selected locations and update their finding location references
- for location in Location.objects.filter(id__in=finding_locations_to_update):
- finding_location = LocationFindingReference.objects.get(location=location, finding__id=finding_id)
- for status in status_list:
- # Set the status attribute based on whether it is enabled in the POST request
- if status in enable:
- # Enable this status
- finding_location.__setattr__(status, True) # noqa: PLC2801
- # If the status is 'Mitigated', record the auditor and audit time
- if status == FindingLocationStatus.Mitigated:
- finding_location.auditor = request.user
- finding_location.audit_time = timezone.now()
- else:
- # Disable this status
- finding_location.__setattr__(status, False) # noqa: PLC2801
- finding_location.save()
+ for location_ref in LocationFindingReference.objects.filter(location__in=finding_locations_to_update, finding__id=finding_id):
+ # Set the status
+ location_ref.set_status(FindingLocationStatus(status), request.user, timezone.now())
# Add a success message after bulk editing endpoints
messages.add_message(
request,
diff --git a/unittests/scans/trivy_operator/compliance_severity.json b/unittests/scans/trivy_operator/compliance_severity.json
new file mode 100644
index 00000000000..20f401af91d
--- /dev/null
+++ b/unittests/scans/trivy_operator/compliance_severity.json
@@ -0,0 +1,78 @@
+{
+ "apiVersion": "aquasecurity.github.io/v1alpha1",
+ "kind": "ClusterComplianceReport",
+ "metadata": {
+ "creationTimestamp": "2024-03-05T10:38:15Z",
+ "generation": 1,
+ "labels": {
+ "app.kubernetes.io/instance": "trivy-operator",
+ "app.kubernetes.io/managed-by": "kubectl",
+ "app.kubernetes.io/name": "trivy-operator",
+ "app.kubernetes.io/version": "0.18.5"
+ },
+ "name": "cis",
+ "resourceVersion": "1649372",
+ "uid": "test-compliance-severity"
+ },
+ "spec": {
+ "compliance": {
+ "controls": [],
+ "description": "Test Compliance Severity",
+ "id": "test",
+ "title": "Test Compliance Severity Report",
+ "version": "1.0"
+ },
+ "cron": "0 */6 * * *",
+ "reportType": "all"
+ },
+ "status": {
+ "detailReport": {
+ "description": "Test report for compliance severity logic",
+ "id": "test",
+ "relatedResources": [],
+ "results": [
+ {
+ "checks": [
+ {
+ "category": "Kubernetes Security Check",
+ "checkID": "AVD-KSV-0001",
+ "description": "Check with its own severity",
+ "messages": [
+ "Test message 1"
+ ],
+ "remediation": "Fix it",
+ "severity": "MEDIUM",
+ "success": false,
+ "target": "/test-target-1",
+ "title": "Check with severity"
+ },
+ {
+ "category": "Kubernetes Security Check",
+ "checkID": "AVD-KSV-0002",
+ "description": "Check without severity",
+ "messages": [
+ "Test message 2"
+ ],
+ "remediation": "Fix it too",
+ "severity": "",
+ "success": false,
+ "target": "/test-target-2",
+ "title": "Check without severity"
+ }
+ ],
+ "description": "Test result",
+ "id": "1.1.1",
+ "name": "Test result name",
+ "severity": "HIGH"
+ }
+ ],
+ "title": "Test Compliance Severity Report",
+ "version": "1.0"
+ },
+ "summary": {
+ "failCount": 2,
+ "passCount": 0
+ },
+ "updateTimestamp": "2024-03-05T10:38:15Z"
+ }
+}
diff --git a/unittests/scans/trivy_operator/configauditreport_missing_checkid.json b/unittests/scans/trivy_operator/configauditreport_missing_checkid.json
new file mode 100644
index 00000000000..67066415b20
--- /dev/null
+++ b/unittests/scans/trivy_operator/configauditreport_missing_checkid.json
@@ -0,0 +1,48 @@
+{
+ "apiVersion": "aquasecurity.github.io/v1alpha1",
+ "kind": "ConfigAuditReport",
+ "metadata": {
+ "annotations": {
+ "trivy-operator.aquasecurity.github.io/report-ttl": "24h0m0s"
+ },
+ "creationTimestamp": "2023-03-23T16:22:54Z",
+ "generation": 1,
+ "labels": {
+ "plugin-config-hash": "659b7b9c46",
+ "resource-spec-hash": "fc85b485f",
+ "trivy-operator.resource.kind": "ReplicaSet",
+ "trivy-operator.resource.name": "test-deployment-12345",
+ "trivy-operator.resource.namespace": "default"
+ },
+ "name": "replicaset-test-deployment-12345",
+ "namespace": "default",
+ "resourceVersion": "1268",
+ "uid": "test-missing-checkid"
+ },
+ "report": {
+ "checks": [
+ {
+ "category": "Kubernetes Security Check",
+ "description": "A check without a checkID field",
+ "messages": [
+ "Container 'test' of ReplicaSet 'test-deployment-12345' has an issue"
+ ],
+ "severity": "MEDIUM",
+ "success": false,
+ "title": "Missing checkID test"
+ }
+ ],
+ "scanner": {
+ "name": "Trivy",
+ "vendor": "Aqua Security",
+ "version": "dev"
+ },
+ "summary": {
+ "criticalCount": 0,
+ "highCount": 0,
+ "lowCount": 0,
+ "mediumCount": 1
+ },
+ "updateTimestamp": "2023-03-23T16:22:54Z"
+ }
+}
diff --git a/unittests/test_import_reimport.py b/unittests/test_import_reimport.py
index 4952854de9e..417fe0ea9ea 100644
--- a/unittests/test_import_reimport.py
+++ b/unittests/test_import_reimport.py
@@ -11,6 +11,7 @@
from django.test.client import Client
from django.urls import reverse
from django.utils import timezone
+from parameterized import parameterized
from dojo.models import Engagement, Finding, Product, Product_Type, Test, Test_Type, User
@@ -2003,6 +2004,74 @@ def test_import_nuclei_emptyc(self):
test_id2 = reimport0["test"]
self.assertEqual(test_id, test_id2)
+ @parameterized.expand(
+ [
+ ("Test False Positive Status (Endpoint Status)", {"false_positive": True}, "status_finding"),
+ ("Test Out of Scope Status (Endpoint Status)", {"out_of_scope": True}, "status_finding"),
+ ("Test Risk Accepted Status (Endpoint Status)", {"risk_accepted": True}, "status_finding"),
+ ("Test False Positive Status (Locations)", {"status": "FalsePositive"}, "locations"),
+ ("Test Out of Scope Status (Locations)", {"status": "OutOfScope"}, "locations"),
+ ("Test Risk Accepted Status (Locations)", {"status": "RiskAccepted"}, "locations"),
+ ],
+ )
+ def test_import_reimport_endpoint_where_eps_reactivation_skips_special_status(self, label: str, special_status_fields: dict, m2m_key: str):
+ """
+ When Findings are set to False Positive, Out of Scope, or Risk Accepted, they are not reactivated
+ because these statuses are often set by humans. The same needs to apply for the Endpoint Status as
+ they are an extension of the finding being partially mitigated.
+ """
+ if settings.V3_FEATURE_LOCATIONS:
+ # TODO: Delete this after the move to Locations
+ if m2m_key == "status_finding":
+ # This test will fail for endpoint statuses with locations enabled
+ # return early here
+ return
+ context = {
+ "auditor": User.objects.get(username="admin"),
+ "audit_time": timezone.now(),
+ }
+ # TODO: Delete this after the move to Locations
+ else:
+ if m2m_key == "locations":
+ # This test will fail for locations with locations disabled
+ # return early here
+ return
+ context = {
+ "mitigated": True,
+ "mitigated_by": User.objects.get(username="admin"),
+ "mitigated_time": timezone.now(),
+ }
+ # Now start the test
+ with assertTestImportModelsCreated(self, imports=1, affected_findings=1, created=1):
+ import0 = self.import_scan_with_params(
+ self.gitlab_dast_file_name, self.scan_type_gitlab_dast, active=True, verified=True,
+ )
+ test_id = import0["test"]
+ findings = self.get_test_findings_api(test_id)
+ self.assert_finding_count_json(1, findings)
+ finding = Finding.objects.get(id=findings["results"][0]["id"])
+ # Get the related objects on the finding
+ related_obects = getattr(finding, m2m_key).all()
+ self.assertEqual(len(related_obects), 1)
+ # Update the related objects with the special status fields
+ related_objects_context = {**context, **special_status_fields}
+ related_obects.update(**related_objects_context)
+ # Reimport the same file
+ reimport0 = self.reimport_scan_with_params(
+ test_id, self.gitlab_dast_file_name, scan_type=self.scan_type_gitlab_dast,
+ )
+ test_id = reimport0["test"]
+ findings = self.get_test_findings_api(test_id)
+ self.assert_finding_count_json(1, findings)
+ finding = Finding.objects.get(id=findings["results"][0]["id"])
+ # Get the related objects on the finding
+ related_obects = getattr(finding, m2m_key).all()
+ self.assertEqual(len(related_obects), 1)
+ related_object = related_obects.first()
+ # Ensure the status is the same as the baseline
+ for key, value in related_objects_context.items():
+ self.assertEqual(getattr(related_object, key), value)
+
def test_import_reimport_endpoint_where_eps_date_is_different(self):
endpoint_count_before = self.db_endpoint_count()
endpoint_status_count_before_active = self.db_endpoint_status_count(mitigated=False)
@@ -2514,6 +2583,63 @@ def test_reimport_set_scan_date_parser_sets_date(self):
date = findings["results"][0]["date"]
self.assertEqual(date, "2006-12-26")
+ def test_reimport_auto_create_does_not_close_findings_in_existing_test(self):
+ """
+ Regression test for #14363: when reimport with auto_create_context=True creates
+ a brand new test, close_old_findings must not close findings from other tests in
+ the same engagement scope.
+
+ The serializer now forces close_old_findings=False when calling DefaultImporter
+ in this path. Without the fix, all 4 findings from the pre-existing test would be
+ incorrectly closed.
+ """
+ product_type, _ = Product_Type.objects.get_or_create(name="PT CloseOld AutoCreate")
+ product, _ = Product.objects.get_or_create(
+ name="P CloseOld AutoCreate",
+ description="test",
+ prod_type=product_type,
+ )
+ engagement = Engagement.objects.create(
+ name="E CloseOld AutoCreate",
+ product=product,
+ target_start=timezone.now(),
+ target_end=timezone.now(),
+ )
+
+ acunetix_many_findings = get_unit_tests_scans_path("acunetix") / "many_findings.xml"
+
+ # Step 1: import 4 findings into an existing test (test1) in the engagement.
+ # minimum_severity="Info" is required to include all 4 findings in the file.
+ import1 = self.import_scan_with_params(
+ acunetix_many_findings,
+ scan_type=self.scan_type_acunetix,
+ engagement=engagement.id,
+ minimum_severity="Info",
+ )
+ test1_id = import1["test"]
+ self.assert_finding_count_json(4, self.get_test_findings_api(test1_id, active=True))
+
+ # Step 2: call the reimport endpoint with auto_create_context=True and a
+ # different test_title so a new test is created. close_old_findings=True
+ # is the value a caller would pass (and the reimport default); the serializer
+ # must suppress it when auto-creating a new test. The scan uses a different
+ # file so its hash codes don't overlap with test1's findings, meaning the
+ # bug would close all 4 of test1's findings if the fix were reverted.
+ self.reimport_scan_with_params(
+ None,
+ self.acunetix_file_name,
+ scan_type=self.scan_type_acunetix,
+ test_title="Brand New Test From Reimport",
+ product_name="P CloseOld AutoCreate",
+ engagement_name="E CloseOld AutoCreate",
+ product_type_name="PT CloseOld AutoCreate",
+ auto_create_context=True,
+ close_old_findings=True,
+ )
+
+ # Step 3: test1's 4 findings must all still be active
+ self.assert_finding_count_json(4, self.get_test_findings_api(test1_id, active=True))
+
@override_settings(
IMPORT_REIMPORT_DEDUPE_BATCH_SIZE=200,
IMPORT_REIMPORT_MATCH_BATCH_SIZE=200,
diff --git a/unittests/test_importers_performance.py b/unittests/test_importers_performance.py
index d0bd84101f5..565e04f98e6 100644
--- a/unittests/test_importers_performance.py
+++ b/unittests/test_importers_performance.py
@@ -270,9 +270,9 @@ def test_import_reimport_reimport_performance_pghistory_async(self):
self._import_reimport_performance(
expected_num_queries1=295,
expected_num_async_tasks1=6,
- expected_num_queries2=227,
+ expected_num_queries2=226,
expected_num_async_tasks2=17,
- expected_num_queries3=109,
+ expected_num_queries3=108,
expected_num_async_tasks3=16,
)
@@ -292,9 +292,9 @@ def test_import_reimport_reimport_performance_pghistory_no_async(self):
self._import_reimport_performance(
expected_num_queries1=302,
expected_num_async_tasks1=6,
- expected_num_queries2=234,
+ expected_num_queries2=233,
expected_num_async_tasks2=17,
- expected_num_queries3=116,
+ expected_num_queries3=115,
expected_num_async_tasks3=16,
)
@@ -315,9 +315,9 @@ def test_import_reimport_reimport_performance_pghistory_no_async_with_product_gr
self._import_reimport_performance(
expected_num_queries1=309,
expected_num_async_tasks1=8,
- expected_num_queries2=241,
+ expected_num_queries2=240,
expected_num_async_tasks2=19,
- expected_num_queries3=120,
+ expected_num_queries3=119,
expected_num_async_tasks3=18,
)
diff --git a/unittests/test_permissions_audit.py b/unittests/test_permissions_audit.py
new file mode 100644
index 00000000000..7ef32860be1
--- /dev/null
+++ b/unittests/test_permissions_audit.py
@@ -0,0 +1,1019 @@
+"""
+Security-focused permission tests for the permissions audit.
+
+Tests verify:
+1. Risk Acceptance data is not exposed to users without Risk_Acceptance permission
+2. Metadata batch operations enforce permissions on parent objects
+3. Note removal verifies note-finding relationship (regression)
+4. Benchmark IDOR: update_benchmark rejects bench_id from different product
+5. Object/tool_product parent mismatch returns 403
+6. Risk Acceptance cross-engagement IDOR (H1 #3577434 / #3569882)
+7. Engagement Presets cross-product IDOR (H1 #3577398 / #3570349)
+8. Questionnaire cross-engagement IDOR (H1 #3571957)
+9. Finding Templates exposure via find_template_to_apply (H1 #3577363)
+10. Jira Epic BFLA - Reader cannot trigger update_jira_epic (H1 #3577193)
+"""
+import datetime
+
+from django.test import Client
+from django.urls import reverse
+from django.utils import timezone
+from rest_framework.authtoken.models import Token
+from rest_framework.test import APIClient
+
+from dojo.models import (
+ Answered_Survey,
+ Benchmark_Category,
+ Benchmark_Product,
+ Benchmark_Product_Summary,
+ Benchmark_Requirement,
+ Benchmark_Type,
+ Dojo_User,
+ DojoMeta,
+ Engagement,
+ Engagement_Presets,
+ Engagement_Survey,
+ Finding,
+ Finding_Template,
+ Notes,
+ Objects_Product,
+ Objects_Review,
+ Product,
+ Product_Member,
+ Product_Type,
+ Risk_Acceptance,
+ Role,
+ Test,
+ Test_Type,
+ Tool_Configuration,
+ Tool_Product_Settings,
+ Tool_Type,
+)
+
+from .dojo_test_case import DojoTestCase
+
+
+class TestRiskAcceptanceExposure(DojoTestCase):
+
+ """FindingSerializer must not expose accepted_risks to users without Risk_Acceptance permission."""
+
+ @classmethod
+ def setUpTestData(cls):
+ cls.reader_role = Role.objects.get(name="Reader")
+ cls.writer_role = Role.objects.get(name="Writer")
+
+ # Create product type and product
+ cls.product_type = Product_Type.objects.create(name="RA Exposure Test PT")
+ cls.product = Product.objects.create(
+ name="RA Exposure Test Product",
+ description="Test",
+ prod_type=cls.product_type,
+ )
+
+ # Create users
+ cls.reader_user = Dojo_User.objects.create_user(
+ username="ra_test_reader",
+ password="testTEST1234!@#$", # noqa: S106
+ is_active=True,
+ )
+ cls.writer_user = Dojo_User.objects.create_user(
+ username="ra_test_writer",
+ password="testTEST1234!@#$", # noqa: S106
+ is_active=True,
+ )
+
+ # Assign roles
+ Product_Member.objects.create(
+ product=cls.product,
+ user=cls.reader_user,
+ role=cls.reader_role,
+ )
+ Product_Member.objects.create(
+ product=cls.product,
+ user=cls.writer_user,
+ role=cls.writer_role,
+ )
+
+ # Create engagement, test, finding
+ cls.engagement = Engagement.objects.create(
+ name="RA Test Engagement",
+ product=cls.product,
+ target_start=datetime.date(2024, 1, 1),
+ target_end=datetime.date(2024, 12, 31),
+ )
+ test_type, _ = Test_Type.objects.get_or_create(name="Manual Code Review")
+ cls.test = Test.objects.create(
+ engagement=cls.engagement,
+ test_type=test_type,
+ target_start=timezone.now(),
+ target_end=timezone.now(),
+ )
+ cls.finding = Finding.objects.create(
+ title="RA Test Finding",
+ test=cls.test,
+ severity="High",
+ numerical_severity="S1",
+ reporter=cls.writer_user,
+ )
+
+ # Create risk acceptance linked to the finding
+ cls.risk_acceptance = Risk_Acceptance.objects.create(
+ name="Test RA",
+ owner=cls.writer_user,
+ )
+ cls.risk_acceptance.accepted_findings.add(cls.finding)
+ cls.engagement.risk_acceptance.add(cls.risk_acceptance)
+
+ def _get_finding_as_user(self, user):
+ token, _ = Token.objects.get_or_create(user=user)
+ client = APIClient()
+ client.credentials(HTTP_AUTHORIZATION="Token " + token.key)
+ return client.get(reverse("finding-detail", args=(self.finding.id,)))
+
+ def test_reader_cannot_see_accepted_risks(self):
+ """Reader role lacks Risk_Acceptance permission — accepted_risks must be empty."""
+ response = self._get_finding_as_user(self.reader_user)
+ self.assertEqual(response.status_code, 200)
+ self.assertEqual(response.json()["accepted_risks"], [])
+
+ def test_writer_can_see_accepted_risks(self):
+ """Writer role has Risk_Acceptance permission — accepted_risks must contain data."""
+ response = self._get_finding_as_user(self.writer_user)
+ self.assertEqual(response.status_code, 200)
+ accepted = response.json()["accepted_risks"]
+ self.assertGreater(len(accepted), 0)
+ self.assertEqual(accepted[0]["name"], "Test RA")
+
+
+class TestMetadataBatchPermissions(DojoTestCase):
+
+ """Metadata batch endpoint must enforce permissions on parent objects."""
+
+ @classmethod
+ def setUpTestData(cls):
+ cls.reader_role = Role.objects.get(name="Reader")
+ cls.writer_role = Role.objects.get(name="Writer")
+
+ # Product the user CAN access
+ cls.product_type = Product_Type.objects.create(name="Meta Batch Test PT")
+ cls.accessible_product = Product.objects.create(
+ name="Meta Batch Accessible Product",
+ description="Test",
+ prod_type=cls.product_type,
+ )
+
+ # Product the user CANNOT access
+ cls.inaccessible_product = Product.objects.create(
+ name="Meta Batch Inaccessible Product",
+ description="Test",
+ prod_type=cls.product_type,
+ )
+
+ # User with Writer on accessible product, no role on inaccessible product
+ cls.writer_user = Dojo_User.objects.create_user(
+ username="meta_batch_writer",
+ password="testTEST1234!@#$", # noqa: S106
+ is_active=True,
+ )
+ Product_Member.objects.create(
+ product=cls.accessible_product,
+ user=cls.writer_user,
+ role=cls.writer_role,
+ )
+
+ # User with Reader on accessible product (Reader lacks Product_Edit)
+ cls.reader_user = Dojo_User.objects.create_user(
+ username="meta_batch_reader",
+ password="testTEST1234!@#$", # noqa: S106
+ is_active=True,
+ )
+ Product_Member.objects.create(
+ product=cls.accessible_product,
+ user=cls.reader_user,
+ role=cls.reader_role,
+ )
+
+ def _client_for_user(self, user):
+ token, _ = Token.objects.get_or_create(user=user)
+ client = APIClient()
+ client.credentials(HTTP_AUTHORIZATION="Token " + token.key)
+ return client
+
+ def test_batch_post_unauthorized_product(self):
+ """Writer should be denied when targeting a product they have no access to."""
+ client = self._client_for_user(self.writer_user)
+ response = client.post(
+ reverse("metadata-batch"),
+ data={
+ "product": self.inaccessible_product.id,
+ "metadata": [{"name": "hack_key", "value": "hack_val"}],
+ },
+ format="json",
+ )
+ self.assertIn(response.status_code, [403, 404])
+ self.assertFalse(
+ DojoMeta.objects.filter(
+ product=self.inaccessible_product, name="hack_key",
+ ).exists(),
+ )
+
+ def test_batch_post_reader_cannot_edit(self):
+ """Reader lacks Product_Edit — batch POST should be denied."""
+ client = self._client_for_user(self.reader_user)
+ response = client.post(
+ reverse("metadata-batch"),
+ data={
+ "product": self.accessible_product.id,
+ "metadata": [{"name": "reader_key", "value": "reader_val"}],
+ },
+ format="json",
+ )
+ self.assertIn(response.status_code, [403, 404])
+ self.assertFalse(
+ DojoMeta.objects.filter(
+ product=self.accessible_product, name="reader_key",
+ ).exists(),
+ )
+
+
+class TestNoteRelationshipVerification(DojoTestCase):
+
+ """Regression: remove_note must verify the note belongs to the finding."""
+
+ @classmethod
+ def setUpTestData(cls):
+ cls.owner_role = Role.objects.get(name="Owner")
+
+ cls.product_type = Product_Type.objects.create(name="Note Test PT")
+ cls.product = Product.objects.create(
+ name="Note Test Product",
+ description="Test",
+ prod_type=cls.product_type,
+ )
+
+ cls.user = Dojo_User.objects.create_user(
+ username="note_test_owner",
+ password="testTEST1234!@#$", # noqa: S106
+ is_active=True,
+ )
+ Product_Member.objects.create(
+ product=cls.product,
+ user=cls.user,
+ role=cls.owner_role,
+ )
+
+ cls.engagement = Engagement.objects.create(
+ name="Note Test Engagement",
+ product=cls.product,
+ target_start=datetime.date(2024, 1, 1),
+ target_end=datetime.date(2024, 12, 31),
+ )
+ test_type, _ = Test_Type.objects.get_or_create(name="Manual Code Review")
+ cls.test = Test.objects.create(
+ engagement=cls.engagement,
+ test_type=test_type,
+ target_start=timezone.now(),
+ target_end=timezone.now(),
+ )
+
+ # Create two findings
+ cls.finding_a = Finding.objects.create(
+ title="Note Test Finding A",
+ test=cls.test,
+ severity="High",
+ numerical_severity="S1",
+ reporter=cls.user,
+ )
+ cls.finding_b = Finding.objects.create(
+ title="Note Test Finding B",
+ test=cls.test,
+ severity="Medium",
+ numerical_severity="S2",
+ reporter=cls.user,
+ )
+
+ # Create a note on finding A
+ cls.note = Notes.objects.create(
+ entry="Test note on finding A",
+ author=cls.user,
+ )
+ cls.finding_a.notes.add(cls.note)
+
+ def test_remove_note_from_wrong_finding(self):
+ """Removing a note via a different finding's endpoint must fail."""
+ token, _ = Token.objects.get_or_create(user=self.user)
+ client = APIClient()
+ client.credentials(HTTP_AUTHORIZATION="Token " + token.key)
+
+ response = client.patch(
+ reverse("finding-remove-note", args=(self.finding_b.id,)),
+ data={"note_id": self.note.id},
+ format="json",
+ )
+ self.assertEqual(response.status_code, 400)
+ # Note should still exist
+ self.assertTrue(Notes.objects.filter(id=self.note.id).exists())
+
+ def test_remove_note_from_correct_finding(self):
+ """Removing a note from the correct finding must succeed for the author."""
+ # Create a fresh note so we don't affect other tests
+ note = Notes.objects.create(
+ entry="Disposable test note",
+ author=self.user,
+ )
+ self.finding_a.notes.add(note)
+
+ token, _ = Token.objects.get_or_create(user=self.user)
+ client = APIClient()
+ client.credentials(HTTP_AUTHORIZATION="Token " + token.key)
+
+ response = client.patch(
+ reverse("finding-remove-note", args=(self.finding_a.id,)),
+ data={"note_id": note.id},
+ format="json",
+ )
+ self.assertEqual(response.status_code, 204)
+ self.assertFalse(Notes.objects.filter(id=note.id).exists())
+
+
+class TestBenchmarkIDOR(DojoTestCase):
+
+ """update_benchmark must reject bench_id belonging to a different product."""
+
+ @classmethod
+ def setUpTestData(cls):
+ cls.owner_role = Role.objects.get(name="Owner")
+ cls.product_type = Product_Type.objects.create(name="Bench IDOR Test PT")
+
+ # Two separate products
+ cls.product_a = Product.objects.create(
+ name="Bench IDOR Product A",
+ description="Test",
+ prod_type=cls.product_type,
+ )
+ cls.product_b = Product.objects.create(
+ name="Bench IDOR Product B",
+ description="Test",
+ prod_type=cls.product_type,
+ )
+
+ # User with Owner on both products
+ cls.user = Dojo_User.objects.create_user(
+ username="bench_idor_owner",
+ password="testTEST1234!@#$", # noqa: S106
+ is_active=True,
+ )
+ Product_Member.objects.create(
+ product=cls.product_a, user=cls.user, role=cls.owner_role,
+ )
+ Product_Member.objects.create(
+ product=cls.product_b, user=cls.user, role=cls.owner_role,
+ )
+
+ # Create benchmark type, category, requirement
+ cls.bench_type = Benchmark_Type.objects.create(
+ name="IDOR Test Type", enabled=True,
+ )
+ cls.bench_category = Benchmark_Category.objects.create(
+ type=cls.bench_type, name="V1: Test Category", enabled=True,
+ )
+ cls.bench_requirement = Benchmark_Requirement.objects.create(
+ category=cls.bench_category,
+ objective_number="1.1",
+ objective="Test objective",
+ enabled=True,
+ )
+
+ # Create a benchmark entry for product A
+ cls.bench_product_a = Benchmark_Product.objects.create(
+ product=cls.product_a,
+ control=cls.bench_requirement,
+ )
+
+ # Create benchmark summary for product B (needed for URL)
+ cls.bench_summary_a = Benchmark_Product_Summary.objects.create(
+ product=cls.product_a, benchmark_type=cls.bench_type,
+ )
+ cls.bench_summary_b = Benchmark_Product_Summary.objects.create(
+ product=cls.product_b, benchmark_type=cls.bench_type,
+ )
+
+ def test_update_benchmark_cross_product_rejected(self):
+ """POSTing a bench_id from product A via product B's URL must be denied."""
+ client = Client()
+ client.login(username="bench_idor_owner", password="testTEST1234!@#$") # noqa: S106
+
+ # Try to update product A's benchmark through product B's endpoint
+ url = reverse(
+ "update_product_benchmark",
+ args=(self.product_b.id, self.bench_type.id),
+ )
+ response = client.post(url, {
+ "bench_id": self.bench_product_a.id,
+ "field": "pass_fail",
+ "value": "true",
+ })
+ # Scoped get_object_or_404 returns 404 for cross-product access;
+ # PermissionDenied would give 400/403 via custom handler403 (DD bug)
+ self.assertIn(response.status_code, [400, 403, 404])
+
+ def test_update_benchmark_summary_cross_product_rejected(self):
+ """POSTing a summary from product A via product B's URL must be denied."""
+ client = Client()
+ client.login(username="bench_idor_owner", password="testTEST1234!@#$") # noqa: S106
+
+ url = reverse(
+ "update_product_benchmark_summary",
+ args=(self.product_b.id, self.bench_type.id, self.bench_summary_a.id),
+ )
+ response = client.post(url, {
+ "field": "publish",
+ "value": "true",
+ })
+ # Scoped get_object_or_404 returns 404 for cross-product access;
+ # PermissionDenied would give 400/403 via custom handler403 (DD bug)
+ self.assertIn(response.status_code, [400, 403, 404])
+
+ def test_update_benchmark_same_product_allowed(self):
+ """POSTing a bench_id for the correct product should succeed."""
+ client = Client()
+ client.login(username="bench_idor_owner", password="testTEST1234!@#$") # noqa: S106
+
+ url = reverse(
+ "update_product_benchmark",
+ args=(self.product_a.id, self.bench_type.id),
+ )
+ response = client.post(url, {
+ "bench_id": self.bench_product_a.id,
+ "field": "enabled",
+ "value": "true",
+ })
+ self.assertEqual(response.status_code, 200)
+
+
+class TestObjectProductParentCheck(DojoTestCase):
+
+ """edit_object and delete_object must reject objects from different products."""
+
+ @classmethod
+ def setUpTestData(cls):
+ cls.owner_role = Role.objects.get(name="Owner")
+ cls.product_type = Product_Type.objects.create(name="Object Parent Test PT")
+
+ cls.product_a = Product.objects.create(
+ name="Object Parent Product A",
+ description="Test",
+ prod_type=cls.product_type,
+ )
+ cls.product_b = Product.objects.create(
+ name="Object Parent Product B",
+ description="Test",
+ prod_type=cls.product_type,
+ )
+
+ cls.user = Dojo_User.objects.create_user(
+ username="object_parent_owner",
+ password="testTEST1234!@#$", # noqa: S106
+ is_active=True,
+ )
+ Product_Member.objects.create(
+ product=cls.product_a, user=cls.user, role=cls.owner_role,
+ )
+ Product_Member.objects.create(
+ product=cls.product_b, user=cls.user, role=cls.owner_role,
+ )
+
+ # Object belonging to product A
+ cls.review_status = Objects_Review.objects.create(name="In Review")
+ cls.tracked_file = Objects_Product.objects.create(
+ product=cls.product_a,
+ path="/test/path",
+ folder="test_folder",
+ artifact="test.py",
+ review_status=cls.review_status,
+ )
+
+ def test_edit_object_cross_product_rejected(self):
+ """Editing an object from product A via product B's URL must be denied."""
+ client = Client()
+ client.login(username="object_parent_owner", password="testTEST1234!@#$") # noqa: S106
+
+ url = reverse("edit_object", args=(self.product_b.id, self.tracked_file.id))
+ response = client.get(url)
+ # PermissionDenied raised; custom handler403 returns 400 (DD bug)
+ self.assertIn(response.status_code, [400, 403])
+
+ def test_delete_object_cross_product_rejected(self):
+ """Deleting an object from product A via product B's URL must be denied."""
+ client = Client()
+ client.login(username="object_parent_owner", password="testTEST1234!@#$") # noqa: S106
+
+ url = reverse("delete_object", args=(self.product_b.id, self.tracked_file.id))
+ response = client.get(url)
+ # PermissionDenied raised; custom handler403 returns 400 (DD bug)
+ self.assertIn(response.status_code, [400, 403])
+
+
+class TestToolProductParentCheck(DojoTestCase):
+
+ """edit_tool_product and delete_tool_product must reject tools from different products."""
+
+ @classmethod
+ def setUpTestData(cls):
+ cls.owner_role = Role.objects.get(name="Owner")
+ cls.product_type = Product_Type.objects.create(name="Tool Parent Test PT")
+
+ cls.product_a = Product.objects.create(
+ name="Tool Parent Product A",
+ description="Test",
+ prod_type=cls.product_type,
+ )
+ cls.product_b = Product.objects.create(
+ name="Tool Parent Product B",
+ description="Test",
+ prod_type=cls.product_type,
+ )
+
+ cls.user = Dojo_User.objects.create_user(
+ username="tool_parent_owner",
+ password="testTEST1234!@#$", # noqa: S106
+ is_active=True,
+ )
+ Product_Member.objects.create(
+ product=cls.product_a, user=cls.user, role=cls.owner_role,
+ )
+ Product_Member.objects.create(
+ product=cls.product_b, user=cls.user, role=cls.owner_role,
+ )
+
+ # Tool type, configuration, and tool setting belonging to product A
+ cls.tool_type = Tool_Type.objects.create(name="Test Tool Type Parent Check")
+ cls.tool_config = Tool_Configuration.objects.create(
+ name="Test Tool Config",
+ tool_type=cls.tool_type,
+ )
+ cls.tool_setting = Tool_Product_Settings.objects.create(
+ name="Test Tool Setting",
+ product=cls.product_a,
+ tool_configuration=cls.tool_config,
+ )
+
+ def test_edit_tool_product_cross_product_rejected(self):
+ """Editing a tool setting from product A via product B's URL must be denied."""
+ client = Client()
+ client.login(username="tool_parent_owner", password="testTEST1234!@#$") # noqa: S106
+
+ url = reverse("edit_tool_product", args=(self.product_b.id, self.tool_setting.id))
+ response = client.get(url)
+ # PermissionDenied raised; custom handler403 returns 400 (DD bug)
+ self.assertIn(response.status_code, [400, 403])
+
+ def test_delete_tool_product_cross_product_rejected(self):
+ """Deleting a tool setting from product A via product B's URL must be denied."""
+ client = Client()
+ client.login(username="tool_parent_owner", password="testTEST1234!@#$") # noqa: S106
+
+ url = reverse("delete_tool_product", args=(self.product_b.id, self.tool_setting.id))
+ response = client.get(url)
+ # PermissionDenied raised; custom handler403 returns 400 (DD bug)
+ self.assertIn(response.status_code, [400, 403])
+
+
+class TestRiskAcceptanceCrossEngagementIDOR(DojoTestCase):
+
+ """
+ H1 #3577434 / #3569882: Risk acceptance endpoints must reject
+ a raid belonging to a different engagement than the eid in the URL.
+ """
+
+ @classmethod
+ def setUpTestData(cls):
+ cls.owner_role = Role.objects.get(name="Owner")
+ cls.product_type = Product_Type.objects.create(name="RA IDOR Test PT")
+ cls.product = Product.objects.create(
+ name="RA IDOR Test Product",
+ description="Test",
+ prod_type=cls.product_type,
+ )
+ cls.user = Dojo_User.objects.create_user(
+ username="ra_idor_owner",
+ password="testTEST1234!@#$", # noqa: S106
+ is_active=True,
+ )
+ Product_Member.objects.create(
+ product=cls.product, user=cls.user, role=cls.owner_role,
+ )
+
+ # Two engagements under the same product
+ cls.engagement_a = Engagement.objects.create(
+ name="RA IDOR Engagement A",
+ product=cls.product,
+ target_start=datetime.date(2024, 1, 1),
+ target_end=datetime.date(2024, 12, 31),
+ )
+ cls.engagement_b = Engagement.objects.create(
+ name="RA IDOR Engagement B",
+ product=cls.product,
+ target_start=datetime.date(2024, 1, 1),
+ target_end=datetime.date(2024, 12, 31),
+ )
+
+ # Create a risk acceptance on engagement A
+ test_type, _ = Test_Type.objects.get_or_create(name="Manual Code Review")
+ cls.test_a = Test.objects.create(
+ engagement=cls.engagement_a,
+ test_type=test_type,
+ target_start=timezone.now(),
+ target_end=timezone.now(),
+ )
+ cls.finding_a = Finding.objects.create(
+ title="RA IDOR Finding",
+ test=cls.test_a,
+ severity="High",
+ numerical_severity="S1",
+ reporter=cls.user,
+ )
+ cls.risk_acceptance = Risk_Acceptance.objects.create(
+ name="RA IDOR Test RA",
+ owner=cls.user,
+ )
+ cls.risk_acceptance.accepted_findings.add(cls.finding_a)
+ cls.engagement_a.risk_acceptance.add(cls.risk_acceptance)
+
+ def _login(self):
+ client = Client()
+ client.login(username="ra_idor_owner", password="testTEST1234!@#$") # noqa: S106
+ return client
+
+ def test_view_risk_acceptance_cross_engagement(self):
+ """Viewing a risk acceptance via a different engagement's URL must be denied."""
+ client = self._login()
+ url = reverse("view_risk_acceptance", args=(
+ self.engagement_b.id, self.risk_acceptance.id,
+ ))
+ response = client.get(url)
+ self.assertEqual(response.status_code, 404)
+
+ def test_edit_risk_acceptance_cross_engagement(self):
+ """Editing a risk acceptance via a different engagement's URL must be denied."""
+ client = self._login()
+ url = reverse("edit_risk_acceptance", args=(
+ self.engagement_b.id, self.risk_acceptance.id,
+ ))
+ response = client.get(url)
+ self.assertEqual(response.status_code, 404)
+
+ def test_expire_risk_acceptance_cross_engagement(self):
+ """Expiring a risk acceptance via a different engagement's URL must be denied."""
+ client = self._login()
+ url = reverse("expire_risk_acceptance", args=(
+ self.engagement_b.id, self.risk_acceptance.id,
+ ))
+ response = client.get(url)
+ self.assertEqual(response.status_code, 404)
+
+ def test_reinstate_risk_acceptance_cross_engagement(self):
+ """Reinstating a risk acceptance via a different engagement's URL must be denied."""
+ client = self._login()
+ url = reverse("reinstate_risk_acceptance", args=(
+ self.engagement_b.id, self.risk_acceptance.id,
+ ))
+ response = client.get(url)
+ self.assertEqual(response.status_code, 404)
+
+ def test_delete_risk_acceptance_cross_engagement(self):
+ """Deleting a risk acceptance via a different engagement's URL must be denied."""
+ client = self._login()
+ url = reverse("delete_risk_acceptance", args=(
+ self.engagement_b.id, self.risk_acceptance.id,
+ ))
+ response = client.get(url)
+ self.assertEqual(response.status_code, 404)
+
+ def test_view_risk_acceptance_same_engagement(self):
+ """Viewing a risk acceptance via the correct engagement's URL should work."""
+ client = self._login()
+ url = reverse("view_risk_acceptance", args=(
+ self.engagement_a.id, self.risk_acceptance.id,
+ ))
+ response = client.get(url)
+ self.assertEqual(response.status_code, 200)
+
+
+class TestEngagementPresetsCrossProductIDOR(DojoTestCase):
+
+ """
+ H1 #3577398 / #3570349: Engagement preset endpoints must reject
+ a preset belonging to a different product than the pid in the URL.
+ """
+
+ @classmethod
+ def setUpTestData(cls):
+ cls.owner_role = Role.objects.get(name="Owner")
+ cls.product_type = Product_Type.objects.create(name="Preset IDOR Test PT")
+
+ cls.product_a = Product.objects.create(
+ name="Preset IDOR Product A",
+ description="Test",
+ prod_type=cls.product_type,
+ )
+ cls.product_b = Product.objects.create(
+ name="Preset IDOR Product B",
+ description="Test",
+ prod_type=cls.product_type,
+ )
+
+ cls.user = Dojo_User.objects.create_user(
+ username="preset_idor_owner",
+ password="testTEST1234!@#$", # noqa: S106
+ is_active=True,
+ )
+ Product_Member.objects.create(
+ product=cls.product_a, user=cls.user, role=cls.owner_role,
+ )
+ Product_Member.objects.create(
+ product=cls.product_b, user=cls.user, role=cls.owner_role,
+ )
+
+ # Preset belonging to product A
+ cls.preset = Engagement_Presets.objects.create(
+ title="IDOR Test Preset",
+ product=cls.product_a,
+ scope="Test scope",
+ )
+
+ def _login(self):
+ client = Client()
+ client.login(username="preset_idor_owner", password="testTEST1234!@#$") # noqa: S106
+ return client
+
+ def test_edit_preset_cross_product(self):
+ """Editing a preset from product A via product B's URL must return 404."""
+ client = self._login()
+ url = reverse("edit_engagement_presets", args=(
+ self.product_b.id, self.preset.id,
+ ))
+ response = client.get(url)
+ # Scoped get_object_or_404 returns 404 for cross-product access
+ self.assertEqual(response.status_code, 404)
+
+ def test_delete_preset_cross_product(self):
+ """Deleting a preset from product A via product B's URL must return 404."""
+ client = self._login()
+ url = reverse("delete_engagement_presets", args=(
+ self.product_b.id, self.preset.id,
+ ))
+ response = client.get(url)
+ self.assertEqual(response.status_code, 404)
+
+ def test_edit_preset_same_product(self):
+ """Editing a preset via the correct product's URL should work."""
+ client = self._login()
+ url = reverse("edit_engagement_presets", args=(
+ self.product_a.id, self.preset.id,
+ ))
+ response = client.get(url)
+ self.assertEqual(response.status_code, 200)
+
+
+class TestQuestionnaireCrossEngagementIDOR(DojoTestCase):
+
+ """
+ H1 #3571957: Survey/questionnaire endpoints must reject
+ a survey belonging to a different engagement than the eid in the URL.
+ """
+
+ @classmethod
+ def setUpTestData(cls):
+ cls.owner_role = Role.objects.get(name="Owner")
+ cls.product_type = Product_Type.objects.create(name="Survey IDOR Test PT")
+ cls.product = Product.objects.create(
+ name="Survey IDOR Test Product",
+ description="Test",
+ prod_type=cls.product_type,
+ )
+ cls.user = Dojo_User.objects.create_user(
+ username="survey_idor_owner",
+ password="testTEST1234!@#$", # noqa: S106
+ is_active=True,
+ )
+ Product_Member.objects.create(
+ product=cls.product, user=cls.user, role=cls.owner_role,
+ )
+
+ cls.engagement_a = Engagement.objects.create(
+ name="Survey IDOR Engagement A",
+ product=cls.product,
+ target_start=datetime.date(2024, 1, 1),
+ target_end=datetime.date(2024, 12, 31),
+ )
+ cls.engagement_b = Engagement.objects.create(
+ name="Survey IDOR Engagement B",
+ product=cls.product,
+ target_start=datetime.date(2024, 1, 1),
+ target_end=datetime.date(2024, 12, 31),
+ )
+
+ # Create a questionnaire (Engagement_Survey) and an Answered_Survey on engagement A
+ cls.survey_template = Engagement_Survey.objects.create(
+ name="Test Questionnaire",
+ description="Test description",
+ active=True,
+ )
+ cls.answered_survey = Answered_Survey.objects.create(
+ engagement=cls.engagement_a,
+ survey=cls.survey_template,
+ responder=cls.user,
+ completed=False,
+ )
+
+ def _login(self):
+ client = Client()
+ client.login(username="survey_idor_owner", password="testTEST1234!@#$") # noqa: S106
+ return client
+
+ def test_view_questionnaire_cross_engagement(self):
+ """Viewing a survey from engagement A via engagement B's URL must return 404."""
+ client = self._login()
+ url = reverse("view_questionnaire", args=(
+ self.engagement_b.id, self.answered_survey.id,
+ ))
+ response = client.get(url)
+ self.assertEqual(response.status_code, 404)
+
+ def test_delete_survey_cross_engagement(self):
+ """Deleting a survey from engagement A via engagement B's URL must return 404."""
+ client = self._login()
+ url = reverse("delete_engagement_survey", args=(
+ self.engagement_b.id, self.answered_survey.id,
+ ))
+ response = client.get(url)
+ self.assertEqual(response.status_code, 404)
+
+ def test_answer_questionnaire_cross_engagement(self):
+ """Answering a survey from engagement A via engagement B's URL must return 404."""
+ client = self._login()
+ url = reverse("answer_questionnaire", args=(
+ self.engagement_b.id, self.answered_survey.id,
+ ))
+ response = client.get(url)
+ self.assertEqual(response.status_code, 404)
+
+ def test_view_questionnaire_same_engagement(self):
+ """Viewing a survey via the correct engagement's URL should work."""
+ client = self._login()
+ url = reverse("view_questionnaire", args=(
+ self.engagement_a.id, self.answered_survey.id,
+ ))
+ response = client.get(url)
+ self.assertEqual(response.status_code, 200)
+
+
+class TestFindingTemplatesGlobalPermission(DojoTestCase):
+
+ """
+ H1 #3577363: find_template_to_apply must require global Finding_Edit
+ permission, not just product-level Finding_Edit.
+ """
+
+ @classmethod
+ def setUpTestData(cls):
+ cls.writer_role = Role.objects.get(name="Writer")
+ cls.product_type = Product_Type.objects.create(name="Template Test PT")
+ cls.product = Product.objects.create(
+ name="Template Test Product",
+ description="Test",
+ prod_type=cls.product_type,
+ )
+
+ # Product-level writer (no global permission)
+ cls.product_writer = Dojo_User.objects.create_user(
+ username="template_test_writer",
+ password="testTEST1234!@#$", # noqa: S106
+ is_active=True,
+ )
+ Product_Member.objects.create(
+ product=cls.product, user=cls.product_writer, role=cls.writer_role,
+ )
+
+ # Superuser (has global permissions)
+ cls.superuser = Dojo_User.objects.create_user(
+ username="template_test_super",
+ password="testTEST1234!@#$", # noqa: S106
+ is_active=True,
+ is_superuser=True,
+ )
+
+ # Create engagement, test, finding
+ cls.engagement = Engagement.objects.create(
+ name="Template Test Engagement",
+ product=cls.product,
+ target_start=datetime.date(2024, 1, 1),
+ target_end=datetime.date(2024, 12, 31),
+ )
+ test_type, _ = Test_Type.objects.get_or_create(name="Manual Code Review")
+ cls.test = Test.objects.create(
+ engagement=cls.engagement,
+ test_type=test_type,
+ target_start=timezone.now(),
+ target_end=timezone.now(),
+ )
+ cls.finding = Finding.objects.create(
+ title="Template Test Finding",
+ test=cls.test,
+ severity="High",
+ numerical_severity="S1",
+ reporter=cls.product_writer,
+ )
+
+ # Create a template (should only be visible to global permission holders)
+ Finding_Template.objects.create(
+ title="Secret Template",
+ severity="Critical",
+ )
+
+ def test_product_writer_cannot_access_find_template(self):
+ """Product-level Writer without global permission should be denied."""
+ client = Client()
+ client.login(username="template_test_writer", password="testTEST1234!@#$") # noqa: S106
+ url = reverse("find_template_to_apply", args=(self.finding.id,))
+ response = client.get(url)
+ # PermissionDenied raised; custom handler403 returns 400 (DD bug)
+ self.assertIn(response.status_code, [400, 403])
+
+ def test_superuser_can_access_find_template(self):
+ """Superuser (implicit global permission) should be able to access."""
+ client = Client()
+ client.login(username="template_test_super", password="testTEST1234!@#$") # noqa: S106
+ url = reverse("find_template_to_apply", args=(self.finding.id,))
+ response = client.get(url)
+ self.assertEqual(response.status_code, 200)
+
+
+class TestJiraEpicBFLA(DojoTestCase):
+
+ """
+ H1 #3577193: update_jira_epic must enforce Engagement_Edit permission,
+ not just IsAuthenticated. Reader role should be denied.
+ """
+
+ @classmethod
+ def setUpTestData(cls):
+ cls.reader_role = Role.objects.get(name="Reader")
+ cls.writer_role = Role.objects.get(name="Writer")
+ cls.product_type = Product_Type.objects.create(name="Jira Epic BFLA Test PT")
+ cls.product = Product.objects.create(
+ name="Jira Epic BFLA Test Product",
+ description="Test",
+ prod_type=cls.product_type,
+ )
+
+ cls.reader_user = Dojo_User.objects.create_user(
+ username="jira_epic_reader",
+ password="testTEST1234!@#$", # noqa: S106
+ is_active=True,
+ )
+ cls.writer_user = Dojo_User.objects.create_user(
+ username="jira_epic_writer",
+ password="testTEST1234!@#$", # noqa: S106
+ is_active=True,
+ )
+
+ Product_Member.objects.create(
+ product=cls.product, user=cls.reader_user, role=cls.reader_role,
+ )
+ Product_Member.objects.create(
+ product=cls.product, user=cls.writer_user, role=cls.writer_role,
+ )
+
+ cls.engagement = Engagement.objects.create(
+ name="Jira Epic BFLA Engagement",
+ product=cls.product,
+ target_start=datetime.date(2024, 1, 1),
+ target_end=datetime.date(2024, 12, 31),
+ )
+
+ def _client_for_user(self, user):
+ token, _ = Token.objects.get_or_create(user=user)
+ client = APIClient()
+ client.credentials(HTTP_AUTHORIZATION="Token " + token.key)
+ return client
+
+ def test_reader_cannot_update_jira_epic(self):
+ """Reader role should be denied POST to update_jira_epic."""
+ client = self._client_for_user(self.reader_user)
+ url = reverse("engagement-update-jira-epic", args=(self.engagement.id,))
+ response = client.post(url, data={}, format="json")
+ self.assertIn(response.status_code, [403, 404])
+
+ def test_writer_allowed_update_jira_epic(self):
+ """
+ Writer role should be allowed to POST to update_jira_epic
+ (may fail at Jira level, but not at permission level).
+ """
+ client = self._client_for_user(self.writer_user)
+ url = reverse("engagement-update-jira-epic", args=(self.engagement.id,))
+ response = client.post(url, data={}, format="json")
+ # Writer has Engagement_Edit, so should pass permission check.
+ # May get 400/500 from Jira integration, but NOT 403.
+ self.assertNotEqual(response.status_code, 403)
diff --git a/unittests/test_rest_framework.py b/unittests/test_rest_framework.py
index 5666f4de1d8..5f10e4e1ad8 100644
--- a/unittests/test_rest_framework.py
+++ b/unittests/test_rest_framework.py
@@ -3788,26 +3788,101 @@ def __init__(self, *args, **kwargs):
def __del__(self: object):
self.payload["file"].close()
+ def _build_payload(self, data):
+ return {
+ "product": 1,
+ "file": SimpleUploadedFile(
+ "defectdojo_cloc.json",
+ json.dumps(data).encode("utf-8"),
+ content_type="application/json",
+ ),
+ }
+
def test_create(self):
- BaseClass.CreateRequestTest.test_create(self)
+ self.payload["file"].close()
+ base_data = json.loads(
+ Path("unittests/files/defectdojo_cloc.json").read_text(
+ encoding="utf-8",
+ ),
+ )
+ updated_data = json.loads(json.dumps(base_data))
+ updated_data.pop("JSON", None)
+ updated_data["Python"]["code"] = 51057
+ updated_data["Go"] = {
+ "nFiles": 1,
+ "blank": 2,
+ "comment": 3,
+ "code": 4,
+ }
- languages = Languages.objects.filter(product=1).order_by("language")
+ test_cases = [
+ (
+ "initial",
+ base_data,
+ {
+ "JSON": {
+ "files": 21,
+ "blank": 7,
+ "comment": 0,
+ "code": 63996,
+ },
+ "Python": {
+ "files": 432,
+ "blank": 10813,
+ "comment": 5054,
+ "code": 51056,
+ },
+ },
+ ),
+ (
+ "updated",
+ updated_data,
+ {
+ "Go": {
+ "files": 1,
+ "blank": 2,
+ "comment": 3,
+ "code": 4,
+ },
+ "Python": {
+ "files": 432,
+ "blank": 10813,
+ "comment": 5054,
+ "code": 51057,
+ },
+ },
+ ),
+ ]
- self.assertEqual(2, len(languages))
+ product = Product.objects.get(id=1)
+ for case_name, payload_data, expected in test_cases:
+ with self.subTest(case=case_name):
+ self.payload = self._build_payload(payload_data)
+ response = self.client.post(self.url, self.payload)
+ self.assertEqual(201, response.status_code, response.content[:1000])
+ self.check_schema_response("post", "201", response)
- self.assertEqual(languages[0].product, Product.objects.get(id=1))
- self.assertEqual(languages[0].language, Language_Type.objects.get(id=1))
- self.assertEqual(languages[0].files, 21)
- self.assertEqual(languages[0].blank, 7)
- self.assertEqual(languages[0].comment, 0)
- self.assertEqual(languages[0].code, 63996)
+ languages = (
+ Languages.objects.filter(product=1)
+ .select_related("language")
+ .order_by("language__language")
+ )
+ self.assertEqual(len(expected), languages.count())
- self.assertEqual(languages[1].product, Product.objects.get(id=1))
- self.assertEqual(languages[1].language, Language_Type.objects.get(id=2))
- self.assertEqual(languages[1].files, 432)
- self.assertEqual(languages[1].blank, 10813)
- self.assertEqual(languages[1].comment, 5054)
- self.assertEqual(languages[1].code, 51056)
+ languages_by_name = {
+ language.language.language: language
+ for language in languages
+ }
+ self.assertEqual(set(expected.keys()), set(languages_by_name.keys()))
+
+ for name, counts in expected.items():
+ language = languages_by_name[name]
+ self.assertEqual(product, language.product)
+ self.assertEqual(name, language.language.language)
+ self.assertEqual(counts["files"], language.files)
+ self.assertEqual(counts["blank"], language.blank)
+ self.assertEqual(counts["comment"], language.comment)
+ self.assertEqual(counts["code"], language.code)
@versioned_fixtures
diff --git a/unittests/tools/test_trivy_operator_parser.py b/unittests/tools/test_trivy_operator_parser.py
index 33181a053c8..b99b99fb8fa 100644
--- a/unittests/tools/test_trivy_operator_parser.py
+++ b/unittests/tools/test_trivy_operator_parser.py
@@ -142,7 +142,7 @@ def test_cis_benchmark(self):
self.assertEqual(len(findings), 795)
finding = findings[0]
self.assertEqual("5.1.2 AVD-KSV-0041 /clusterrole-admin", finding.title)
- self.assertEqual("High", finding.severity)
+ self.assertEqual("Critical", finding.severity)
self.assertEqual(1, len(finding.unsaved_vulnerability_ids))
self.assertEqual("AVD-KSV-0041", finding.unsaved_vulnerability_ids[0])
finding = findings[40]
@@ -174,6 +174,27 @@ def test_findings_clustercompliancereport(self):
findings = parser.get_findings(test_file, Test())
self.assertEqual(len(findings), 2)
+ def test_compliance_severity_logic(self):
+ with sample_path("compliance_severity.json").open(encoding="utf-8") as test_file:
+ parser = TrivyOperatorParser()
+ findings = parser.get_findings(test_file, Test())
+ self.assertEqual(len(findings), 2)
+ # First check has severity MEDIUM, result has severity HIGH -> uses check's MEDIUM
+ self.assertEqual("Medium", findings[0].severity)
+ # Second check has empty severity, result has severity HIGH -> falls back to HIGH
+ self.assertEqual("High", findings[1].severity)
+
+ def test_configauditreport_missing_checkid(self):
+ with sample_path("configauditreport_missing_checkid.json").open(encoding="utf-8") as test_file:
+ parser = TrivyOperatorParser()
+ findings = parser.get_findings(test_file, Test())
+ self.assertEqual(len(findings), 1)
+ finding = findings[0]
+ self.assertEqual("Medium", finding.severity)
+ self.assertEqual("0 - Missing checkID test", finding.title)
+ # When checkID is "0", references should be empty (not a bogus URL)
+ self.assertEqual("", finding.references)
+
def test_configauditreport_with_remediation(self):
with sample_path("configauditreport_with_remediation.json").open(encoding="utf-8") as test_file:
parser = TrivyOperatorParser()