wip

Author: valentijnscholten

dojo/jira_link/helper.py

@@ -119,27 +119,25 @@ def _safely_get_obj_status_for_jira(obj: Finding | Finding_Group, *, isenforced:
             return obj.status
 
     if isinstance(obj, Finding_Group):
-        jira_minimum_threshold = 0
-        if System_Settings.objects.get().jira_minimum_severity:
-            jira_minimum_threshold = Finding.get_numerical_severity(System_Settings.objects.get().jira_minimum_severity)
-
-        findings = obj.findings.all().filter(numerical_severity__gte=jira_minimum_threshold)
+        findings = get_finding_group_findings_above_threshold(obj)
         if not findings:
             return ["Empty"]
 
+        for find in findings:
+            logger.error(f"Finding {find.id} status {find.active} {find.verified} {find.is_mitigated}")
+
         # This iterates 3 times over the list of findings, but any code doing 1 iteration would looke it's from 1990
-        has_active = any(find.active for find in findings)
-        has_verified = any(find.verified for find in findings)
-        all_mitigated = all(find.is_mitigated for find in findings) if findings else False
+        if any(find.active for find in findings):
+            status += ["Active"]
+
+        if any((find.active and find.verified) for find in findings):
+            status += ["Verified"]
 
-        if has_active:
-            status.append("Active")
-        if has_verified:
-            status.append("Verified")
-        if all_mitigated:
-            status.append("Mitigated")
+        if all(find.is_mitigated for find in findings):
+            status += ["Mitigated", "Inactive"]
 
-    return status
+    # if no active findings are found, we must assume the status is inactive
+    return status or ["Inactive"]
 
 
 # checks if a finding can be pushed to JIRA
@@ -193,6 +191,7 @@ def can_be_pushed_to_jira(obj, form=None):
             return False, f"Finding below the minimum JIRA severity threshold ({System_Settings.objects.get().jira_minimum_severity}).", "error_below_minimum_threshold"
     elif isinstance(obj, Finding_Group):
         finding_group_status = _safely_get_obj_status_for_jira(obj)
+        logger.error(f"Finding group status: {finding_group_status}")
         if "Empty" in finding_group_status:
             return False, f"{to_str_typed(obj)} cannot be pushed to jira as it contains no findings above minimum treshold.", "error_empty"
 
@@ -673,7 +672,17 @@ def jira_description(obj):
 
 
 def jira_priority(obj):
-    return get_jira_instance(obj).get_priority(obj.severity)
+    if isinstance(obj, Finding):
+        return get_jira_instance(obj).get_priority(obj.severity)
+    if isinstance(obj, Finding_Group):
+        finding_group_severity_for_jira = get_finding_group_findings_above_threshold(obj)
+
+        max_number_severity = max(Finding.get_number_severity(find.severity) for find in finding_group_severity_for_jira)
+        return Finding.get_severity(max_number_severity)
+
+    logger.error("unsupported object passed to push_to_jira: %s %i %s", obj.__name__, obj.id, obj)
+    msg = f"Unsupported object passed to push_to_jira: {type(obj)}"
+    raise RuntimeError(msg)
 
 
 def jira_environment(obj):
@@ -1122,8 +1131,7 @@ def push_status_to_jira(obj, jira_instance, jira, issue, *, save=False):
     status_list = _safely_get_obj_status_for_jira(obj)
     issue_closed = False
     updated = False
-    logger.debug("item check: %s", "Active" in "Active")  # noqa: PLR0133
-    logger.debug("any item resolved: %s", any(item in status_list for item in RESOLVED_STATUS))
+    logger.debug("pushing status to JIRA for %d:%s status:%s", obj.id, to_str_typed(obj), status_list)
     # check RESOLVED_STATUS first to avoid corner cases with findings that are Inactive, but verified
     if not updated and any(item in status_list for item in RESOLVED_STATUS):
         if issue_from_jira_is_active(issue):
@@ -1777,3 +1785,16 @@ def save_and_push_to_jira(finding):
     # the updated data of the finding is pushed as part of the group
     if push_to_jira_decision and finding_in_group:
         push_to_jira(finding.finding_group)
+
+
+def get_finding_group_findings_above_threshold(finding_group):
+    """Get the findings that are above the minimum threshold"""
+    jira_minimum_threshold = 0
+    if System_Settings.objects.get().jira_minimum_severity:
+        jira_minimum_threshold = Finding.get_numerical_severity(System_Settings.objects.get().jira_minimum_severity)
+
+    findings = finding_group.findings.filter(numerical_severity__lte=jira_minimum_threshold)
+    # TODO: JIRA REMOVE
+    logger.error(findings.query)
+    logger.error(f"count: {findings.count()}")
+    return findings

dojo/templates/issue-trackers/jira_full/jira-finding-group-description.tpl

@@ -10,12 +10,14 @@ A group of Findings has been pushed to JIRA to be investigated and fixed:
 h2. Group
 *Group*: [{{ finding_group.name|jiraencode}}|{{ finding_group_url|full_url }}] in [{{ finding_group.test.engagement.product.name|jiraencode }}|{{ product_url|full_url }}] / [{{ finding_group.test.engagement.name|jiraencode }}|{{ engagement_url|full_url }}] / [{{ finding_group.test|stringformat:'s'|jiraencode }}|{{ test_url|full_url }}]
 
-
+# TODO: JIRA only include active/verified findings above threshold
 || Severity || CVE || CWE || Component || Version || Title || Status ||{% for finding in finding_group.findings.all %}
 | {{finding.severity}} | {% if finding.cve %}[{{finding.cve}}|{{finding.cve|vulnerability_url}}]{% else %}None{% endif %} | [{{finding.cwe}}|{{finding.cwe|cwe_url}}] | {{finding.component_name|jiraencode_component}} | {{finding.component_version}} | {% url 'view_finding' finding.id as finding_url %}[{{ finding.title|jiraencode}}|{{ finding_url|full_url }}] | {{ finding.status }} |{% endfor %}
 
+# TODO: JIRA only include active/verified findings above threshold
 *Severity:* {{ finding_group.severity }}
 
+# TODO: JIRA only include active/verified findings above threshold
 {% if finding_group.sla_deadline %} *Due Date:* {{ finding_group.sla_deadline }} {% endif %}
 
 {% if finding_group.test.engagement.branch_tag %}

run-unittest.sh

@@ -1,4 +1,5 @@
 #!/usr/bin/env bash
+set -x
 
 unset TEST_CASE
 
@@ -51,7 +52,7 @@ then
 fi
 
 echo "Running docker compose unit tests with test case $TEST_CASE ..."
-# Compose V2 integrates compose functions into the Docker platform, continuing to support 
-# most of the  previous docker-compose features and flags. You can run Compose V2 by 
+# Compose V2 integrates compose functions into the Docker platform, continuing to support
+# most of the  previous docker-compose features and flags. You can run Compose V2 by
 # replacing the hyphen (-) with a space, using docker compose, instead of docker-compose.
-docker compose exec uwsgi bash -c "python manage.py test $TEST_CASE -v2 --keepdb"
+docker compose exec -e JIRA_PAT_DD=$JIRA_PAT_DD=$JIRA_PAT_DD=$JIRA_PAT_DD uwsgi bash -c "python manage.py test $TEST_CASE -v2 --keepdb --failfast"

unittests/dojo_test_case.py

@@ -65,6 +65,26 @@ def wrapper(*args, **kwargs):
     return decorator
 
 
+def with_system_setting(field, value):
+    """Decorator to temporarily set a value in System Settings."""
+
+    def decorator(test_func):
+        @wraps(test_func)
+        def wrapper(*args, **kwargs):
+            old_value = getattr(System_Settings.objects.get(), field)
+            # Set the flag to the specified value
+            System_Settings.objects.update(**{field: value})
+            try:
+                return test_func(*args, **kwargs)
+            finally:
+                # Reset the flag to its original state after the test
+                System_Settings.objects.update(**{field: old_value})
+
+        return wrapper
+
+    return decorator
+
+
 class DojoTestUtilsMixin:
 
     def get_test_admin(self, *args, **kwargs):

unittests/scans/npm_audit/many_vuln_with_groups.json

@@ -188,7 +188,7 @@
             "recommendation": "Update to version 0.6.1 or later.",
             "references": "",
             "access": "public",
-            "severity": "high",
+            "severity": "moderate",
             "cwe": "CWE-400",
             "metadata": {
                 "module_type": "Network.Library",
@@ -317,7 +317,7 @@
             "recommendation": "Update to version 0.5.2 or later.",
             "references": "",
             "access": "public",
-            "severity": "high",
+            "severity": "moderate",
             "cwe": "CWE-400",
             "metadata": {
                 "module_type": "Multi.Library",
@@ -359,7 +359,7 @@
             "recommendation": "* Version 2.x.x: Update to version 2.11.2 or later.\n* Version 3.x.x: Update to version 3.6.4 or later.\n* Version 4.x.x: Update to version 4.5.7 or later.\n* Version 5.x.x: Update to version 5.2.1 or later.\n* Version 6.x.x: Update to version 6.4.2 or later. ( Note that versions 6.1.6, 6.2.5, and 6.3.3 are also patched. )\n* Version 7.x.x: Update to version 7.1.2 or later. ( Note that version 7.0.2 is also patched. )",
             "references": "[Node Postgres: Code Execution Vulnerability Announcement](https://node-postgres.com/announcements#2017-08-12-code-execution-vulnerability)",
             "access": "public",
-            "severity": "high",
+            "severity": "moderate",
             "cwe": "CWE-94",
             "metadata": {
                 "module_type": "Network.Library",

unittests/test_jira_import_and_pushing_api.py

@@ -16,6 +16,7 @@
     get_unit_tests_path,
     get_unit_tests_scans_path,
     toggle_system_setting_boolean,
+    with_system_setting,
 )
 
 logger = logging.getLogger(__name__)
@@ -49,12 +50,12 @@ def __init__(self, *args, **kwargs):
         DojoVCRAPITestCase.__init__(self, *args, **kwargs)
 
     def assert_cassette_played(self):
-        if True:  # set to True when committing. set to False when recording new test cassettes
+        if False:  # set to True when committing. set to False when recording new test cassettes
             self.assertTrue(self.cassette.all_played)
 
     def _get_vcr(self, **kwargs):
         my_vcr = super()._get_vcr(**kwargs)
-        my_vcr.record_mode = "once"
+        my_vcr.record_mode = "all"
         my_vcr.path_transformer = VCR.ensure_suffix(".yaml")
         my_vcr.filter_headers = ["Authorization", "X-Atlassian-Token"]
         my_vcr.cassette_library_dir = str(get_unit_tests_path() / "vcr" / "jira")
@@ -69,6 +70,7 @@ def setUp(self):
         self.testuser = User.objects.get(username="admin")
         self.testuser.usercontactinfo.block_execution = True
         self.testuser.usercontactinfo.save()
+
         token = Token.objects.get(user=self.testuser)
         self.client = APIClient()
         self.client.credentials(HTTP_AUTHORIZATION="Token " + token.key)
@@ -106,6 +108,29 @@ def test_import_with_groups_push_to_jira(self):
         # by asserting full cassette is played we know issues have been updated in JIRA
         self.assert_cassette_played()
 
+    @with_system_setting("jira_minimum_severity", "Critical")
+    def test_import_with_groups_push_to_jira_minimum_critical(self):
+        # No Critical findings in report, so expect no groups to be pushed
+        import0 = self.import_scan_with_params(self.npm_groups_sample_filename, scan_type="NPM Audit Scan", group_by="component_name+component_version", push_to_jira=True, verified=True)
+        test_id = import0["test"]
+        # all findings should be in a group, so no JIRA issues for individual findings
+        self.assert_jira_issue_count_in_test(test_id, 0)
+        self.assert_jira_group_issue_count_in_test(test_id, 0)
+        # by asserting full cassette is played we know issues have been updated in JIRA
+        self.assert_cassette_played()
+
+    @with_system_setting("jira_minimum_severity", "High")
+    def test_import_with_groups_push_to_jira_minimum_high(self):
+        # 7 findings, 5 unique component_name+component_version
+        import0 = self.import_scan_with_params(self.npm_groups_sample_filename, scan_type="NPM Audit Scan", group_by="component_name+component_version", push_to_jira=True, verified=True)
+        test_id = import0["test"]
+        # all findings should be in a group, so no JIRA issues for individual findings
+        self.assert_jira_issue_count_in_test(test_id, 0)
+        # fresh library has only medium findings, so only 2 instead of 3 groups expected
+        self.assert_jira_group_issue_count_in_test(test_id, 2)
+        # by asserting full cassette is played we know issues have been updated in JIRA
+        self.assert_cassette_played()
+
     def test_import_with_push_to_jira_epic_as_issue_type(self):
         jira_instance = JIRA_Instance.objects.get(id=2)
         # we choose issue type Epic and test if it can be created successfully.
@@ -456,7 +481,7 @@ def test_groups_create_edit_update_finding(self):
         del finding_details["push_to_jira"]
 
         # push a finding should result in pushing the group instead
-        self.patch_finding_api(findings["results"][0]["id"], {"push_to_jira": True})
+        self.patch_finding_api(findings["results"][0]["id"], {"push_to_jira": True, "verified": True})
 
         self.assert_jira_issue_count_in_test(test_id, 0)
         self.assert_jira_group_issue_count_in_test(test_id, 1)
@@ -628,7 +653,12 @@ def test_import_with_push_to_jira_not_verified_enforced_verified_globally_true_e
         import0 = self.import_scan_with_params(self.zap_sample5_filename, push_to_jira=True, verified=False)
         test_id = import0["test"]
         # This scan file has two active findings, so we should not push either of them
-        self.assert_jira_issue_count_in_test(test_id, 0)
+        self.assert_jira_group_issue_count_in_test(test_id, 0)
+
+        import0 = self.import_scan_with_params(self.zap_sample5_filename, push_to_jira=True, verified=True)
+        test_id = import0["test"]
+        self.assert_jira_group_issue_count_in_test(test_id, 2)
+
         # by asserting full cassette is played we know all calls to JIRA have been made as expected
         self.assert_cassette_played()
 
@@ -639,7 +669,12 @@ def test_import_with_push_to_jira_not_verified_enforced_verified_globally_true_e
         test_id = import0["test"]
         # This scan file has two active findings, so we should not push either of them
         self.assert_jira_issue_count_in_test(test_id, 0)
+
+        import0 = self.import_scan_with_params(self.zap_sample5_filename, push_to_jira=True, verified=True)
+        test_id = import0["test"]
+        self.assert_jira_issue_count_in_test(test_id, 2)
         # by asserting full cassette is played we know all calls to JIRA have been made as expected
+
         self.assert_cassette_played()
 
     @toggle_system_setting_boolean("enforce_verified_status", False)  # noqa: FBT003
@@ -649,6 +684,11 @@ def test_import_with_push_to_jira_not_verified_enforced_verified_globally_false_
         test_id = import0["test"]
         # This scan file has two active findings, so we should not push either of them
         self.assert_jira_issue_count_in_test(test_id, 0)
+
+        import0 = self.import_scan_with_params(self.zap_sample5_filename, push_to_jira=True, verified=True)
+        test_id = import0["test"]
+        self.assert_jira_issue_count_in_test(test_id, 2)
+
         # by asserting full cassette is played we know all calls to JIRA have been made as expected
         self.assert_cassette_played()
 
@@ -662,6 +702,61 @@ def test_import_with_push_to_jira_not_verified_enforced_verified_globally_false_
         # by asserting full cassette is played we know all calls to JIRA have been made as expected
         self.assert_cassette_played()
 
+    @toggle_system_setting_boolean("enforce_verified_status", True)  # noqa: FBT003
+    @toggle_system_setting_boolean("enforce_verified_status_jira", True)  # noqa: FBT003
+    def test_groups_import_with_push_to_jira_not_verified_enforced_verified_globally_true_enforced_verified_jira_true(self):
+        import0 = self.import_scan_with_params(self.npm_groups_sample_filename, scan_type="NPM Audit Scan", group_by="component_name+component_version", push_to_jira=True, verified=False)
+        test_id = import0["test"]
+        # No verified findings, means no groups pushed to JIRA
+        self.assert_jira_group_issue_count_in_test(test_id, 0)
+
+        import0 = self.import_scan_with_params(self.npm_groups_sample_filename, scan_type="NPM Audit Scan", group_by="component_name+component_version", push_to_jira=True, verified=True)
+        test_id = import0["test"]
+        self.assert_jira_group_issue_count_in_test(test_id, 3)
+
+        # by asserting full cassette is played we know all calls to JIRA have been made as expected
+        self.assert_cassette_played()
+
+    @toggle_system_setting_boolean("enforce_verified_status", True)  # noqa: FBT003
+    @toggle_system_setting_boolean("enforce_verified_status_jira", False)  # noqa: FBT003
+    def test_groups_import_with_push_to_jira_not_verified_enforced_verified_globally_true_enforced_verified_jira_false(self):
+        import0 = self.import_scan_with_params(self.npm_groups_sample_filename, scan_type="NPM Audit Scan", group_by="component_name+component_version", push_to_jira=True, verified=False)
+        test_id = import0["test"]
+        # No verified findings, means no groups pushed to JIRA
+        self.assert_jira_group_issue_count_in_test(test_id, 0)
+
+        import0 = self.import_scan_with_params(self.npm_groups_sample_filename, scan_type="NPM Audit Scan", group_by="component_name+component_version", push_to_jira=True, verified=True)
+        test_id = import0["test"]
+        self.assert_jira_group_issue_count_in_test(test_id, 3)
+        # by asserting full cassette is played we know all calls to JIRA have been made as expected
+
+        self.assert_cassette_played()
+
+    @toggle_system_setting_boolean("enforce_verified_status", False)  # noqa: FBT003
+    @toggle_system_setting_boolean("enforce_verified_status_jira", True)  # noqa: FBT003
+    def test_groups_import_with_push_to_jira_not_verified_enforced_verified_globally_false_enforced_verified_jira_true(self):
+        import0 = self.import_scan_with_params(self.npm_groups_sample_filename, scan_type="NPM Audit Scan", group_by="component_name+component_version", push_to_jira=True, verified=False)
+        test_id = import0["test"]
+        # No verified findings, means no groups pushed to JIRA
+        self.assert_jira_group_issue_count_in_test(test_id, 0)
+
+        import0 = self.import_scan_with_params(self.npm_groups_sample_filename, scan_type="NPM Audit Scan", group_by="component_name+component_version", push_to_jira=True, verified=True)
+        test_id = import0["test"]
+        self.assert_jira_group_issue_count_in_test(test_id, 3)
+
+        # by asserting full cassette is played we know all calls to JIRA have been made as expected
+        self.assert_cassette_played()
+
+    @toggle_system_setting_boolean("enforce_verified_status", False)  # noqa: FBT003
+    @toggle_system_setting_boolean("enforce_verified_status_jira", False)  # noqa: FBT003
+    @with_system_setting("jira_minimum_severity", "Low")
+    def test_groups_import_with_push_to_jira_not_verified_enforced_verified_globally_false_enforced_verified_jira_false(self):
+        import0 = self.import_scan_with_params(self.npm_groups_sample_filename, scan_type="NPM Audit Scan", group_by="component_name+component_version", push_to_jira=True, verified=True)
+        test_id = import0["test"]
+        self.assert_jira_group_issue_count_in_test(test_id, 3)
+        # by asserting full cassette is played we know all calls to JIRA have been made as expected
+        self.assert_cassette_played()
+
     def test_engagement_epic_creation(self):
         eng = self.get_engagement(3)
         # Set epic_mapping to true