") and stripped.endswith("
"): + return stripped[3:-4] + return stripped + + +def parse_os_message(text): + lines = text.splitlines() + + headline_source = None + body_start = None + for index, line in enumerate(lines): + if line.startswith("# "): + headline_source = line[2:].strip() + body_start = index + 1 + break + + if not headline_source: + return None + + headline_source = headline_source[:100] + headline_rendered = markdown.markdown(headline_source) + headline_cleaned = bleach.clean( + headline_rendered, + tags=INLINE_TAGS, + attributes=INLINE_ATTRS, + strip=True, + ) + headline_html = _strip_outer_p(headline_cleaned) + + expanded_html = None + expanded_marker = "## Expanded Message" + expanded_body_lines = None + for offset, line in enumerate(lines[body_start:], start=body_start): + if line.strip() == expanded_marker: + expanded_body_lines = lines[offset + 1:] + break + + if expanded_body_lines is not None: + expanded_source = "\n".join(expanded_body_lines).strip() + if expanded_source: + expanded_rendered = markdown.markdown( + expanded_source, + extensions=["extra", "fenced_code", "nl2br"], + ) + expanded_html = bleach.clean( + expanded_rendered, + tags=BLOCK_TAGS, + attributes=BLOCK_ATTRS, + strip=True, + ) + + return {"message": headline_html, "expanded_html": expanded_html} + + +def get_os_banner(): + try: + text = fetch_os_message() + if not text: + return None + return parse_os_message(text) + except Exception: + logger.debug("os_message: get_os_banner failed", exc_info=True) + return None diff --git a/dojo/announcement/signals.py b/dojo/announcement/signals.py index dedd3444654..c74fd0e5d50 100644 --- a/dojo/announcement/signals.py +++ b/dojo/announcement/signals.py @@ -1,4 +1,3 @@ -from django.conf import settings from django.db.models.signals import post_save from django.dispatch import receiver @@ -7,22 +6,11 @@ @receiver(post_save, sender=Dojo_User) def add_announcement_to_new_user(sender, instance, **kwargs): - announcements = Announcement.objects.all() - if announcements.count() > 0: - dojo_user = Dojo_User.objects.get(id=instance.id) - announcement = announcements.first() - cloud_announcement = ( - "DefectDojo Pro Cloud and On-Premise Subscriptions Now Available!" - in announcement.message + announcement = Announcement.objects.first() + if announcement is not None: + UserAnnouncement.objects.get_or_create( + user=instance, announcement=announcement, ) - if not cloud_announcement or settings.CREATE_CLOUD_BANNER: - user_announcements = UserAnnouncement.objects.filter( - user=dojo_user, announcement=announcement, - ) - if user_announcements.count() == 0: - UserAnnouncement.objects.get_or_create( - user=dojo_user, announcement=announcement, - ) @receiver(post_save, sender=Announcement) diff --git a/dojo/api_v2/permissions.py b/dojo/api_v2/permissions.py index 807bbcf7b2a..10e012ede5b 100644 --- a/dojo/api_v2/permissions.py +++ b/dojo/api_v2/permissions.py @@ -473,7 +473,13 @@ def has_permission(self, request, view): # Raise an explicit drf exception here raise ValidationError(e) if engagement := converted_dict.get("engagement"): - # existing engagement, nothing special to check + # Validate the resolved engagement's parent chain matches any provided identifiers + if (product := converted_dict.get("product")) and engagement.product_id != product.id: + msg = "The provided identifiers are inconsistent — the engagement does not belong to the specified product." + raise ValidationError(msg) + if (engagement_name := converted_dict.get("engagement_name")) and engagement.name != engagement_name: + msg = "The provided identifiers are inconsistent — the engagement name does not match the specified engagement." + raise ValidationError(msg) return user_has_permission( request.user, engagement, Permissions.Import_Scan_Result, ) @@ -764,6 +770,11 @@ def has_permission(self, request, view): try: converted_dict = auto_create.convert_querydict_to_dict(request.data) auto_create.process_import_meta_data_from_dict(converted_dict) + # engagement is not a declared field on ReImportScanSerializer and will be + # stripped during validation — don't use it in the permission check either, + # so the permission check resolves targets the same way execution does + converted_dict.pop("engagement", None) + converted_dict.pop("engagement_id", None) # Get an existing product converted_dict["product_type"] = auto_create.get_target_product_type_if_exists(**converted_dict) converted_dict["product"] = auto_create.get_target_product_if_exists(**converted_dict) @@ -774,7 +785,20 @@ def has_permission(self, request, view): raise ValidationError(e) if test := converted_dict.get("test"): - # existing test, nothing special to check + # Validate the resolved test's parent chain matches any provided identifiers + if (product := converted_dict.get("product")) and test.engagement.product_id != product.id: + msg = "The provided identifiers are inconsistent — the test does not belong to the specified product." + raise ValidationError(msg) + if (engagement := converted_dict.get("engagement")) and test.engagement_id != engagement.id: + msg = "The provided identifiers are inconsistent — the test does not belong to the specified engagement." + raise ValidationError(msg) + # Also validate by name when the objects were not resolved (e.g. names that match no existing record) + if not converted_dict.get("product") and (product_name := converted_dict.get("product_name")) and test.engagement.product.name != product_name: + msg = "The provided identifiers are inconsistent — the test does not belong to the specified product." + raise ValidationError(msg) + if not converted_dict.get("engagement") and (engagement_name := converted_dict.get("engagement_name")) and test.engagement.name != engagement_name: + msg = "The provided identifiers are inconsistent — the test does not belong to the specified engagement." + raise ValidationError(msg) return user_has_permission( request.user, test, Permissions.Import_Scan_Result, ) @@ -1181,7 +1205,10 @@ def check_auto_create_permission( raise ValidationError(msg) if engagement: - # existing engagement, nothing special to check + # Validate the resolved engagement's parent chain matches any provided names + if product is not None and engagement.product_id != product.id: + msg = "The provided identifiers are inconsistent — the engagement does not belong to the specified product." + raise ValidationError(msg) return user_has_permission( user, engagement, Permissions.Import_Scan_Result, ) diff --git a/dojo/api_v2/serializers.py b/dojo/api_v2/serializers.py index 342287aba94..0dbd15bf436 100644 --- a/dojo/api_v2/serializers.py +++ b/dojo/api_v2/serializers.py @@ -24,7 +24,6 @@ from rest_framework.fields import DictField, MultipleChoiceField import dojo.finding.helper as finding_helper -import dojo.jira_link.helper as jira_helper import dojo.risk_acceptance.helper as ra_helper from dojo.authorization.authorization import user_has_permission from dojo.authorization.roles_permissions import Permissions @@ -40,6 +39,7 @@ from dojo.importers.base_importer import BaseImporter from dojo.importers.default_importer import DefaultImporter from dojo.importers.default_reimporter import DefaultReImporter +from dojo.jira import services as jira_services from dojo.location.models import Location, LocationFindingReference from dojo.models import ( DEFAULT_NOTIFICATION, @@ -75,9 +75,6 @@ Finding_Template, General_Survey, Global_Role, - JIRA_Instance, - JIRA_Issue, - JIRA_Project, Language_Type, Languages, Network_Locations, @@ -1123,6 +1120,18 @@ def validate(self, data): if data.get("target_start") > data.get("target_end"): msg = "Your target start date exceeds your target end date" raise serializers.ValidationError(msg) + if ( + self.instance is not None + and "product" in data + and data.get("product") != self.instance.product + and not user_has_permission( + self.context["request"].user, + data.get("product"), + Permissions.Engagement_Edit, + ) + ): + msg = "You are not permitted to edit engagements in the destination product" + raise PermissionDenied(msg) return data def build_relational_field(self, field_name, relation_info): @@ -1364,79 +1373,11 @@ class Meta: fields = "__all__" -class JIRAIssueSerializer(serializers.ModelSerializer): - url = serializers.SerializerMethodField(read_only=True) - - class Meta: - model = JIRA_Issue - fields = "__all__" - - def get_url(self, obj) -> str: - return jira_helper.get_jira_issue_url(obj) - - def validate(self, data): - if self.context["request"].method == "PATCH": - engagement = data.get("engagement", self.instance.engagement) - finding = data.get("finding", self.instance.finding) - finding_group = data.get( - "finding_group", self.instance.finding_group, - ) - else: - engagement = data.get("engagement", None) - finding = data.get("finding", None) - finding_group = data.get("finding_group", None) - - if ( - (engagement and not finding and not finding_group) - or (finding and not engagement and not finding_group) - or (finding_group and not engagement and not finding) - ): - pass - else: - msg = "Either engagement or finding or finding_group has to be set." - raise serializers.ValidationError(msg) - - if finding: - if (linked_finding := jira_helper.jira_already_linked(finding, data.get("jira_key"), data.get("jira_id"))) is not None: - msg = "JIRA issue " + data.get("jira_key") + " already linked to " + reverse("view_finding", args=(linked_finding.id,)) - raise serializers.ValidationError(msg) - - return data - - -class JIRAInstanceSerializer(serializers.ModelSerializer): - class Meta: - model = JIRA_Instance - fields = "__all__" - extra_kwargs = { - "password": {"write_only": True}, - } - - -class JIRAProjectSerializer(serializers.ModelSerializer): - class Meta: - model = JIRA_Project - fields = "__all__" - - def validate(self, data): - if self.context["request"].method == "PATCH": - engagement = data.get("engagement", self.instance.engagement) - product = data.get("product", self.instance.product) - else: - engagement = data.get("engagement", None) - product = data.get("product", None) - - if (engagement and product) or (not engagement and not product): - msg = "Either engagement or product has to be set." - raise serializers.ValidationError(msg) - - if "custom_fields" in data and isinstance(data["custom_fields"], str): - try: - data["custom_fields"] = json.loads(data["custom_fields"]) - except json.JSONDecodeError as e: - raise serializers.ValidationError({"custom_fields": f"Invalid JSON: {e}"}) from e - - return data +from dojo.jira.api.serializers import ( # noqa: E402, F401 backward compat + JIRAInstanceSerializer, + JIRAIssueSerializer, + JIRAProjectSerializer, +) class SonarqubeIssueSerializer(serializers.ModelSerializer): @@ -1758,7 +1699,7 @@ def get_test(self, obj): @extend_schema_field(JIRAIssueSerializer) def get_jira(self, obj): - issue = jira_helper.get_jira_issue(obj) + issue = jira_services.get_issue(obj) if issue is None: return None return JIRAIssueSerializer(read_only=True).to_representation(issue) @@ -1832,11 +1773,11 @@ def get_accepted_risks(self, obj): @extend_schema_field(serializers.DateTimeField()) def get_jira_creation(self, obj): - return jira_helper.get_jira_creation(obj) + return jira_services.get_creation(obj) @extend_schema_field(serializers.DateTimeField()) def get_jira_change(self, obj): - return jira_helper.get_jira_change(obj) + return jira_services.get_change(obj) @extend_schema_field(FindingRelatedFieldsSerializer) def get_related_fields(self, obj): @@ -1912,9 +1853,9 @@ def update(self, instance, validated_data): for location_ref in locations: location_ref.location.associate_with_finding(instance) - if push_to_jira or finding_helper.is_keep_in_sync_with_jira(instance): + if push_to_jira or jira_services.is_keep_in_sync(instance): # Push synchronously so that we can see jira errors in real time - success, message = jira_helper.push_to_jira(instance, sync=True) + success, message = jira_services.push(instance, sync=True) if not success: raise serializers.ValidationError(message) @@ -2071,7 +2012,7 @@ def create(self, validated_data): save_vulnerability_ids(new_finding, parsed_vulnerability_ids) if push_to_jira: - jira_helper.push_to_jira(new_finding) + jira_services.push(new_finding) # Create a notification create_notification( @@ -3069,9 +3010,9 @@ class ReportGenerateSerializer(serializers.Serializer): ) -class EngagementUpdateJiraEpicSerializer(serializers.Serializer): - epic_name = serializers.CharField(required=False, max_length=200) - epic_priority = serializers.CharField(required=False, allow_null=True) +from dojo.jira.api.serializers import ( # noqa: E402, F401 backward compat + EngagementUpdateJiraEpicSerializer, +) class TagSerializer(serializers.Serializer): diff --git a/dojo/api_v2/views.py b/dojo/api_v2/views.py index 85627178af9..4da5a02885c 100644 --- a/dojo/api_v2/views.py +++ b/dojo/api_v2/views.py @@ -37,7 +37,6 @@ from rest_framework.response import Response import dojo.finding.helper as finding_helper -import dojo.jira_link.helper as jira_helper from dojo.api_v2 import ( mixins as dojo_mixins, ) @@ -88,10 +87,7 @@ get_authorized_groups, ) from dojo.importers.auto_create_context import AutoCreateContextManager -from dojo.jira_link.queries import ( - get_authorized_jira_issues, - get_authorized_jira_projects, -) +from dojo.jira import services as jira_services from dojo.labels import get_labels from dojo.models import ( Announcement, @@ -117,9 +113,6 @@ Finding_Template, General_Survey, Global_Role, - JIRA_Instance, - JIRA_Issue, - JIRA_Project, Language_Type, Languages, Network_Locations, @@ -720,15 +713,18 @@ def download_file(self, request, file_id, pk=None): def update_jira_epic(self, request, pk=None): engagement = self.get_object() try: - if engagement.has_jira_issue: - dojo_dispatch_task(jira_helper.update_epic, engagement.id, **request.data) + task = jira_services.get_epic_task("update_epic") + if task: + dojo_dispatch_task(task, engagement.id, **request.data) response = Response( {"info": "Jira Epic update query sent"}, status=status.HTTP_200_OK, ) else: - dojo_dispatch_task(jira_helper.add_epic, engagement.id, **request.data) + task = jira_services.get_epic_task("add_epic") + if task: + dojo_dispatch_task(task, engagement.id, **request.data) response = Response( {"info": "Jira Epic create query sent"}, status=status.HTTP_200_OK, @@ -1088,7 +1084,7 @@ class FindingViewSet( def perform_update(self, serializer): # IF JIRA is enabled and this product has a JIRA configuration push_to_jira = serializer.validated_data.get("push_to_jira") - jira_project = jira_helper.get_jira_project(serializer.instance) + jira_project = jira_services.get_project(serializer.instance) if get_system_setting("enable_jira") and jira_project: push_to_jira = push_to_jira or jira_project.push_all_issues @@ -1361,9 +1357,9 @@ def notes(self, request, pk=None): ) if finding.has_jira_issue: - jira_helper.add_comment(finding, note) + jira_services.add_comment(finding, note) elif finding.has_jira_group_issue: - jira_helper.add_comment(finding.finding_group, note) + jira_services.add_comment(finding.finding_group, note) serialized_note = serializers.NoteSerializer( {"author": author, "entry": entry, "private": private}, @@ -1769,74 +1765,11 @@ def metadata(self, request, pk=None): # Authorization: configuration -class JiraInstanceViewSet( - DojoModelViewSet, -): - serializer_class = serializers.JIRAInstanceSerializer - queryset = JIRA_Instance.objects.none() - filter_backends = (DjangoFilterBackend,) - filterset_fields = ["id", "url"] - permission_classes = (permissions.UserHasConfigurationPermissionSuperuser,) - - def get_queryset(self): - return JIRA_Instance.objects.all().order_by("id") - - -# Authorization: object-based -# @extend_schema_view(**schema_with_prefetch()) -# Nested models with prefetch make the response schema too long for Swagger UI -class JiraIssuesViewSet( - PrefetchDojoModelViewSet, -): - serializer_class = serializers.JIRAIssueSerializer - queryset = JIRA_Issue.objects.none() - filter_backends = (DjangoFilterBackend,) - filterset_fields = [ - "id", - "jira_id", - "jira_key", - "finding", - "engagement", - "finding_group", - ] - - permission_classes = ( - IsAuthenticated, - permissions.UserHasJiraIssuePermission, - ) - - def get_queryset(self): - return get_authorized_jira_issues(Permissions.Product_View) - - -# Authorization: object-based -@extend_schema_view(**schema_with_prefetch()) -class JiraProjectViewSet( - PrefetchDojoModelViewSet, -): - serializer_class = serializers.JIRAProjectSerializer - queryset = JIRA_Project.objects.none() - filter_backends = (DjangoFilterBackend,) - filterset_fields = [ - "id", - "jira_instance", - "product", - "engagement", - "enabled", - "component", - "project_key", - "push_all_issues", - "enable_engagement_epic_mapping", - "push_notes", - ] - - permission_classes = ( - IsAuthenticated, - permissions.UserHasJiraProductPermission, - ) - - def get_queryset(self): - return get_authorized_jira_projects(Permissions.Product_View) +from dojo.jira.api.views import ( # noqa: E402, F401 backward compat + JiraInstanceViewSet, + JiraIssuesViewSet, + JiraProjectViewSet, +) # Authorization: superuser @@ -2871,7 +2804,7 @@ def perform_create(self, serializer): push_to_jira = serializer.validated_data.get("push_to_jira") if get_system_setting("enable_jira"): jira_driver = engagement or (product or None) - if jira_project := (jira_helper.get_jira_project(jira_driver) if jira_driver else None): + if jira_project := (jira_services.get_project(jira_driver) if jira_driver else None): push_to_jira = push_to_jira or jira_project.push_all_issues # Add pghistory context for audit trail (adds to existing middleware context). @@ -3029,7 +2962,7 @@ def perform_create(self, serializer): push_to_jira = serializer.validated_data.get("push_to_jira") if get_system_setting("enable_jira"): jira_driver = test or (engagement or (product or None)) - if jira_project := (jira_helper.get_jira_project(jira_driver) if jira_driver else None): + if jira_project := (jira_services.get_project(jira_driver) if jira_driver else None): push_to_jira = push_to_jira or jira_project.push_all_issues logger.debug("push_to_jira: %s", push_to_jira) # Add pghistory context for audit trail (adds to existing middleware context) diff --git a/dojo/context_processors.py b/dojo/context_processors.py index cc53af0f1e0..792e1eb6b42 100644 --- a/dojo/context_processors.py +++ b/dojo/context_processors.py @@ -5,13 +5,14 @@ from django.conf import settings from django.contrib import messages +from dojo.announcement.os_message import get_os_banner from dojo.labels import get_labels from dojo.models import Alerts, System_Settings, UserAnnouncement def globalize_vars(request): # return the value you want as a dictionnary. you may add multiple values in there. - return { + context = { "SHOW_LOGIN_FORM": settings.SHOW_LOGIN_FORM, "FORGOT_PASSWORD": settings.FORGOT_PASSWORD, "FORGOT_USERNAME": settings.FORGOT_USERNAME, @@ -35,11 +36,32 @@ def globalize_vars(request): "DOCUMENTATION_URL": settings.DOCUMENTATION_URL, "API_TOKENS_ENABLED": settings.API_TOKENS_ENABLED, "API_TOKEN_AUTH_ENDPOINT_ENABLED": settings.API_TOKEN_AUTH_ENDPOINT_ENABLED, - "CREATE_CLOUD_BANNER": settings.CREATE_CLOUD_BANNER, + "SHOW_PLG_LINK": True, # V3 Feature Flags "V3_FEATURE_LOCATIONS": settings.V3_FEATURE_LOCATIONS, } + additional_banners = [] + + if (os_banner := get_os_banner()) is not None: + additional_banners.append({ + "source": "os", + "message": os_banner["message"], + "style": "info", + "url": "", + "link_text": "", + "expanded_html": os_banner["expanded_html"], + }) + + if hasattr(request, "session"): + for banner in request.session.pop("_product_banners", []): + additional_banners.append(banner) + + if additional_banners: + context["additional_banners"] = additional_banners + + return context + def bind_system_settings(request): """Load system settings and display warning if there's a database error.""" diff --git a/dojo/engagement/services.py b/dojo/engagement/services.py index 42a7c1c05e4..b78844fc6bc 100644 --- a/dojo/engagement/services.py +++ b/dojo/engagement/services.py @@ -4,8 +4,8 @@ from django.db.models.signals import pre_save from django.dispatch import receiver -import dojo.jira_link.helper as jira_helper from dojo.celery_dispatch import dojo_dispatch_task +from dojo.jira import services as jira_services from dojo.models import Engagement logger = logging.getLogger(__name__) @@ -16,8 +16,10 @@ def close_engagement(eng): eng.status = "Completed" eng.save() - if jira_helper.get_jira_project(eng): - dojo_dispatch_task(jira_helper.close_epic, eng.id, push_to_jira=True) + if jira_services.get_project(eng): + task = jira_services.get_epic_task("close_epic") + if task: + dojo_dispatch_task(task, eng.id, push_to_jira=True) def reopen_engagement(eng): diff --git a/dojo/engagement/views.py b/dojo/engagement/views.py index 1b94ace21f4..34c516bcc7b 100644 --- a/dojo/engagement/views.py +++ b/dojo/engagement/views.py @@ -32,7 +32,6 @@ from openpyxl import Workbook from openpyxl.styles import Font -import dojo.jira_link.helper as jira_helper import dojo.risk_acceptance.helper as ra_helper from dojo.authorization.authorization import user_has_permission_or_403 from dojo.authorization.authorization_decorators import user_is_authorized @@ -74,6 +73,7 @@ ) from dojo.importers.base_importer import BaseImporter from dojo.importers.default_importer import DefaultImporter +from dojo.jira import services as jira_services from dojo.location.models import Location from dojo.location.utils import save_locations_to_add from dojo.models import ( @@ -281,11 +281,17 @@ def edit_engagement(request, eid): if request.method == "POST": form = EngForm(request.POST, instance=engagement, cicd=is_ci_cd, product=engagement.product, user=request.user) - jira_project = jira_helper.get_jira_project(engagement, use_inheritance=False) + jira_project = jira_services.get_project(engagement, use_inheritance=False) if form.is_valid(): # first save engagement details new_status = form.cleaned_data.get("status") + if form.cleaned_data.get("product") != engagement.product: + user_has_permission_or_403( + request.user, + form.cleaned_data.get("product"), + Permissions.Engagement_Edit, + ) engagement.product = form.cleaned_data.get("product") engagement = form.save(commit=False) if (new_status in {"Cancelled", "Completed"}): @@ -301,10 +307,10 @@ def edit_engagement(request, eid): "Engagement updated successfully.", extra_tags="alert-success") - success, jira_project_form = jira_helper.process_jira_project_form(request, instance=jira_project, target="engagement", engagement=engagement, product=engagement.product) + success, jira_project_form = jira_services.process_project_form(request, instance=jira_project, target="engagement", engagement=engagement, product=engagement.product) error = not success - success, jira_epic_form = jira_helper.process_jira_epic_form(request, engagement=engagement) + success, jira_epic_form = jira_services.process_epic_form(request, engagement=engagement) error = error or not success if not error: @@ -321,7 +327,7 @@ def edit_engagement(request, eid): jira_epic_form = None if get_system_setting("enable_jira"): - jira_project = jira_helper.get_jira_project(engagement, use_inheritance=False) + jira_project = jira_services.get_project(engagement, use_inheritance=False) jira_project_form = JIRAProjectForm(instance=jira_project, target="engagement", product=engagement.product) logger.debug("showing jira-epic-form") jira_epic_form = JIRAEngagementForm(instance=engagement) @@ -465,8 +471,8 @@ def get(self, request, eid, *args, **kwargs): network = eng.preset.network_locations.all() system_settings = System_Settings.objects.get() - jissue = jira_helper.get_jira_issue(eng) - jira_project = jira_helper.get_jira_project(eng) + jissue = jira_services.get_issue(eng) + jira_project = jira_services.get_project(eng) try: check = Check_List.objects.get(engagement=eng) @@ -534,8 +540,8 @@ def post(self, request, eid, *args, **kwargs): network = eng.preset.network_locations.all() system_settings = System_Settings.objects.get() - jissue = jira_helper.get_jira_issue(eng) - jira_project = jira_helper.get_jira_project(eng) + jissue = jira_services.get_issue(eng) + jira_project = jira_services.get_project(eng) try: check = Check_List.objects.get(engagement=eng) @@ -796,9 +802,9 @@ def get_jira_form( jira_form = None push_all_jira_issues = False # Determine if jira issues should be pushed automatically - push_all_jira_issues = jira_helper.is_push_all_issues(engagement_or_product) + push_all_jira_issues = jira_services.is_push_all_issues(engagement_or_product) # Only return the form if the jira is enabled on this engagement or product - if jira_helper.get_jira_project(engagement_or_product): + if jira_services.get_project(engagement_or_product): if request.method == "POST": jira_form = JIRAImportScanForm( request.POST, @@ -1195,7 +1201,7 @@ def unlink_jira(request, eid): logger.info("trying to unlink a linked jira epic from engagement %d:%s", eng.id, eng.name) if eng.has_jira_issue: try: - jira_helper.unlink_jira(request, eng) + jira_services.unlink(request, eng) messages.add_message( request, messages.SUCCESS, @@ -1442,9 +1448,10 @@ def view_edit_risk_acceptance(request, eid, raid, *, edit_mode=False): "Since you are not the note's author, it was not deleted.", extra_tags="alert-danger") - if "remove_finding" in request.POST: + if edit_mode and "remove_finding" in request.POST: finding = get_object_or_404( - Finding, pk=request.POST["remove_finding_id"]) + risk_acceptance.accepted_findings, + pk=request.POST["remove_finding_id"]) ra_helper.remove_finding_from_risk_acceptance(request.user, risk_acceptance, finding) diff --git a/dojo/finding/helper.py b/dojo/finding/helper.py index ba11567869b..7ec3ba477ea 100644 --- a/dojo/finding/helper.py +++ b/dojo/finding/helper.py @@ -15,7 +15,6 @@ from django.utils.timezone import is_naive, make_aware, now from fieldsignals import pre_save_changed -import dojo.jira_link.helper as jira_helper import dojo.risk_acceptance.helper as ra_helper from dojo.celery import app from dojo.endpoint.utils import endpoint_get_or_create, save_endpoints_to_add @@ -27,7 +26,7 @@ do_false_positive_history_batch, get_finding_models_for_deduplication, ) -from dojo.jira_link.helper import is_keep_in_sync_with_jira +from dojo.jira import services as jira_services from dojo.location.models import Location from dojo.location.status import FindingLocationStatus from dojo.location.utils import save_locations_to_add @@ -241,10 +240,10 @@ def add_to_finding_group(finding_group, finds): finding_group.findings.add(*available_findings) # Now update the JIRA to add the finding to the finding group - jira_instance = jira_helper.get_jira_instance(finding_group) + jira_instance = jira_services.get_instance(finding_group) if finding_group.has_jira_issue and jira_instance and jira_instance.finding_jira_sync: logger.debug("pushing to jira from finding.finding_bulk_update_all()") - jira_helper.push_to_jira(finding_group) + jira_services.push(finding_group) added = len(available_findings) skipped = len(finds) - added @@ -269,10 +268,10 @@ def remove_from_finding_group(finds): # Now update the JIRA to remove the finding from the finding group for group in affected_groups: - jira_instance = jira_helper.get_jira_instance(group) + jira_instance = jira_services.get_instance(group) if group.has_jira_issue and jira_instance and jira_instance.finding_jira_sync: logger.debug("pushing to jira from finding.finding_bulk_update_all()") - jira_helper.push_to_jira(group) + jira_services.push(group) return affected_groups, removed, skipped @@ -347,10 +346,10 @@ def group_findings_by(finds, finding_group_by_option): # Now update the JIRA to add the finding to the finding group for group in affected_groups: - jira_instance = jira_helper.get_jira_instance(group) + jira_instance = jira_services.get_instance(group) if group.has_jira_issue and jira_instance and jira_instance.finding_jira_sync: logger.debug("pushing to jira from finding.finding_bulk_update_all()") - jira_helper.push_to_jira(group) + jira_services.push(group) return affected_groups, grouped, skipped, groups_created @@ -452,9 +451,9 @@ def post_process_finding_save_internal(finding, dedupe_option=True, rules_option # based on feedback we could introduct another push_group_to_jira boolean everywhere # but what about the push_all boolean? Let's see how this works for now and get some feedback. if finding.has_jira_issue or not finding.finding_group: - jira_helper.push_to_jira(finding) + jira_services.push(finding) elif finding.finding_group: - jira_helper.push_to_jira(finding.finding_group) + jira_services.push(finding.finding_group) @app.task @@ -524,8 +523,8 @@ def post_process_findings_batch( for finding in findings: object_to_push = finding if finding.has_jira_issue or not finding.finding_group else finding.finding_group # Check the push_to_jira flag again to potentially shorty circuit without checking for existing findings - if push_to_jira or is_keep_in_sync_with_jira(object_to_push, prefetched_jira_instance=jira_instance): - jira_helper.push_to_jira(object_to_push) + if push_to_jira or jira_services.is_keep_in_sync(object_to_push, prefetched_jira_instance=jira_instance): + jira_services.push(object_to_push) else: logger.debug("push_to_jira is False, not pushing to JIRA") @@ -1158,20 +1157,20 @@ def _save_finding_with_jira_sync(finding, *, new_note=None): jira_issue_exists = finding.has_jira_issue or ( finding.finding_group and finding.finding_group.has_jira_issue ) - jira_instance = jira_helper.get_jira_instance(finding) - jira_project = jira_helper.get_jira_project(finding) + jira_instance = jira_services.get_instance(finding) + jira_project = jira_services.get_project(finding) if jira_issue_exists: push_to_jira = ( - jira_helper.is_push_all_issues(finding) + jira_services.is_push_all_issues(finding) or (jira_instance and jira_instance.finding_jira_sync) ) if new_note and (getattr(jira_project, "push_notes", False) or push_to_jira) and not finding_in_group: - jira_helper.add_comment(finding, new_note, force_push=True) + jira_services.add_comment(finding, new_note, force_push=True) finding.save(push_to_jira=(push_to_jira and not finding_in_group)) if push_to_jira and finding_in_group: - jira_helper.push_to_jira(finding.finding_group) + jira_services.push(finding.finding_group) def close_finding( diff --git a/dojo/finding/views.py b/dojo/finding/views.py index a4173348b4b..e64bea6a29f 100644 --- a/dojo/finding/views.py +++ b/dojo/finding/views.py @@ -30,7 +30,6 @@ from imagekit.processors import ResizeToFill import dojo.finding.helper as finding_helper -import dojo.jira_link.helper as jira_helper import dojo.risk_acceptance.helper as ra_helper from dojo.authorization.authorization import user_has_global_permission_or_403, user_has_permission_or_403 from dojo.authorization.authorization_decorators import ( @@ -79,6 +78,7 @@ StubFindingForm, TypedNoteForm, ) +from dojo.jira import services as jira_services from dojo.location.status import FindingLocationStatus from dojo.models import ( IMPORT_UNTOUCHED_FINDING, @@ -329,7 +329,7 @@ def get_initial_context(self, request: HttpRequest): user_has_permission_or_403(request.user, product, Permissions.Product_View) context["show_product_column"] = False context["product_tab"] = Product_Tab(product, title="Findings", tab="findings") - context["jira_project"] = jira_helper.get_jira_project(product) + context["jira_project"] = jira_services.get_project(product) if github_config := GITHUB_PKey.objects.filter(product=product).first(): context["github_config"] = github_config.git_conf_id elif engagement_id := self.get_engagement_id(): @@ -337,7 +337,7 @@ def get_initial_context(self, request: HttpRequest): user_has_permission_or_403(request.user, engagement, Permissions.Engagement_View) context["show_product_column"] = False context["product_tab"] = Product_Tab(engagement.product, title=engagement.name, tab="engagements") - context["jira_project"] = jira_helper.get_jira_project(engagement) + context["jira_project"] = jira_services.get_project(engagement) if github_config := GITHUB_PKey.objects.filter(product__engagement=engagement).first(): context["github_config"] = github_config.git_conf_id @@ -592,7 +592,7 @@ def get_jira_data(self, finding: Finding): can_be_pushed_to_jira, can_be_pushed_to_jira_error, error_code, - ) = jira_helper.can_be_pushed_to_jira(finding) + ) = jira_services.can_be_pushed(finding) # Check the error code if error_code: logger.debug(error_code) @@ -647,9 +647,9 @@ def process_form(self, request: HttpRequest, finding: Finding, context: dict): finding.save() # Determine if the note should be sent to jira if finding.has_jira_issue: - jira_helper.add_comment(finding, new_note) + jira_services.add_comment(finding, new_note) elif finding.has_jira_group_issue: - jira_helper.add_comment(finding.finding_group, new_note) + jira_services.add_comment(finding.finding_group, new_note) # Send the notification of the note being added url = request.build_absolute_uri( reverse("view_finding", args=(finding.id,)), @@ -765,9 +765,9 @@ def get_finding_form(self, request: HttpRequest, finding: Finding): def get_jira_form(self, request: HttpRequest, finding: Finding, finding_form: FindingForm = None): # Determine if jira should be used - if (jira_project := jira_helper.get_jira_project(finding)) is not None: + if (jira_project := jira_services.get_project(finding)) is not None: # Determine if push all findings is enabled - push_all_findings = jira_helper.is_push_all_issues(finding) + push_all_findings = jira_services.is_push_all_issues(finding) # Set up the args for the form args = [request.POST] if request.method == "POST" else [] # Set the initial form args @@ -987,8 +987,8 @@ def process_jira_form(self, request: HttpRequest, finding: Finding, context: dic logger.debug(JFORM_PUSH_TO_JIRA_MESSAGE, context["jform"].cleaned_data.get("push_to_jira")) # can't use helper as when push_all_jira_issues is True, the checkbox gets disabled and is always false push_to_jira_checkbox = context["jform"].cleaned_data.get("push_to_jira") - push_all_jira_issues = jira_helper.is_push_all_issues(finding) - push_to_jira = push_all_jira_issues or push_to_jira_checkbox or jira_helper.is_keep_in_sync_with_jira(finding) + push_all_jira_issues = jira_services.is_push_all_issues(finding) + push_to_jira = push_all_jira_issues or push_to_jira_checkbox or jira_services.is_keep_in_sync(finding) logger.debug("push_to_jira: %s", push_to_jira) logger.debug("push_all_jira_issues: %s", push_all_jira_issues) logger.debug("has_jira_group_issue: %s", finding.has_jira_group_issue) @@ -1005,14 +1005,14 @@ def process_jira_form(self, request: HttpRequest, finding: Finding, context: dic which is already checked in the validation of the form """ if not new_jira_issue_key: - jira_helper.finding_unlink_jira(request, finding) + jira_services.unlink_finding(request, finding) jira_message = "Link to JIRA issue removed successfully." elif new_jira_issue_key != finding.jira_issue.jira_key: - jira_helper.finding_unlink_jira(request, finding) - jira_helper.finding_link_jira(request, finding, new_jira_issue_key) + jira_services.unlink_finding(request, finding) + jira_services.link_finding(request, finding, new_jira_issue_key) jira_message = "Changed JIRA link successfully." elif new_jira_issue_key: - jira_helper.finding_link_jira(request, finding, new_jira_issue_key) + jira_services.link_finding(request, finding, new_jira_issue_key) jira_message = "Linked a JIRA issue successfully." # any existing finding should be updated # Determine if a message should be added @@ -1069,7 +1069,7 @@ def process_forms(self, request: HttpRequest, finding: Finding, context: dict): # we only push the group after storing the finding to make sure # the updated data of the finding is pushed as part of the group if push_to_jira and finding.finding_group: - jira_helper.push_to_jira(finding.finding_group) + jira_services.push(finding.finding_group) return request, all_forms_valid @@ -1337,8 +1337,8 @@ def defect_finding_review(request, fid): # Only push if the finding is not in a group if jira_issue_exists: # Determine if any automatic sync should occur - jira_instance = jira_helper.get_jira_instance(finding) - push_to_jira = jira_helper.is_push_all_issues(finding) \ + jira_instance = jira_services.get_instance(finding) + push_to_jira = jira_services.is_push_all_issues(finding) \ or (jira_instance and jira_instance.finding_jira_sync) # Add the closing note if push_to_jira and not finding_in_group: @@ -1346,14 +1346,14 @@ def defect_finding_review(request, fid): new_note.entry += "\nJira issue set to resolved." else: new_note.entry += "\nJira issue re-opened." - jira_helper.add_comment(finding, new_note, force_push=True) + jira_services.add_comment(finding, new_note, force_push=True) # Save the finding finding.save(push_to_jira=(push_to_jira and not finding_in_group)) # we only push the group after saving the finding to make sure # the updated data of the finding is pushed as part of the group if push_to_jira and finding_in_group: - jira_helper.push_to_jira(finding.finding_group) + jira_services.push(finding.finding_group) messages.add_message( request, messages.SUCCESS, "Defect Reviewed", extra_tags="alert-success", @@ -1407,8 +1407,8 @@ def reopen_finding(request, fid): # Clear the risk acceptance, if present ra_helper.risk_unaccept(request.user, finding) finding.save(dedupe_option=False, push_to_jira=False) - if jira_helper.is_push_all_issues(finding) or jira_helper.is_keep_in_sync_with_jira(finding): - jira_helper.push_to_jira(finding) + if jira_services.is_push_all_issues(finding) or jira_services.is_keep_in_sync(finding): + jira_services.push(finding) reopen_external_issue(finding.id, "re-opened by defectdojo", "github") @@ -1609,19 +1609,19 @@ def request_finding_review(request, fid): # Only push if the finding is not in a group if jira_issue_exists: # Determine if any automatic sync should occur - jira_instance = jira_helper.get_jira_instance(finding) - push_to_jira = jira_helper.is_push_all_issues(finding) \ + jira_instance = jira_services.get_instance(finding) + push_to_jira = jira_services.is_push_all_issues(finding) \ or (jira_instance and jira_instance.finding_jira_sync) # Add the closing note if push_to_jira and not finding_in_group: - jira_helper.add_comment(finding, new_note, force_push=True) + jira_services.add_comment(finding, new_note, force_push=True) # Save the finding finding.save(push_to_jira=(push_to_jira and not finding_in_group)) # we only push the group after saving the finding to make sure # the updated data of the finding is pushed as part of the group if push_to_jira and finding_in_group: - jira_helper.push_to_jira(finding.finding_group) + jira_services.push(finding.finding_group) reviewers = Dojo_User.objects.filter(id__in=form.cleaned_data["reviewers"]) reviewers_string = ", ".join([f"{user} ({user.id})" for user in reviewers]) @@ -1704,19 +1704,19 @@ def clear_finding_review(request, fid): # Only push if the finding is not in a group if jira_issue_exists: # Determine if any automatic sync should occur - jira_instance = jira_helper.get_jira_instance(finding) - push_to_jira = jira_helper.is_push_all_issues(finding) \ + jira_instance = jira_services.get_instance(finding) + push_to_jira = jira_services.is_push_all_issues(finding) \ or (jira_instance and jira_instance.finding_jira_sync) # Add the closing note if push_to_jira and not finding_in_group: - jira_helper.add_comment(finding, new_note, force_push=True) + jira_services.add_comment(finding, new_note, force_push=True) # Save the finding finding.save(push_to_jira=(push_to_jira and not finding_in_group)) # we only push the group after saving the finding to make sure # the updated data of the finding is pushed as part of the group if push_to_jira and finding_in_group: - jira_helper.push_to_jira(finding.finding_group) + jira_services.push(finding.finding_group) messages.add_message( request, @@ -2099,9 +2099,9 @@ def promote_to_finding(request, fid): finding = get_object_or_404(Stub_Finding, id=fid) test = finding.test form_error = False - push_all_jira_issues = jira_helper.is_push_all_issues(finding) + push_all_jira_issues = jira_services.is_push_all_issues(finding) jform = None - use_jira = jira_helper.get_jira_project(finding) is not None + use_jira = jira_services.get_project(finding) is not None product_tab = Product_Tab( finding.test.engagement.product, title="Promote Finding", tab="findings", ) @@ -2114,7 +2114,7 @@ def promote_to_finding(request, fid): instance=finding, prefix="jiraform", push_all=push_all_jira_issues, - jira_project=jira_helper.get_jira_project(finding), + jira_project=jira_services.get_project(finding), ) if form.is_valid() and (jform is None or jform.is_valid()): @@ -2166,11 +2166,11 @@ def promote_to_finding(request, fid): """ if not new_jira_issue_key: - jira_helper.finding_unlink_jira(request, new_finding) + jira_services.unlink_finding(request, new_finding) elif new_jira_issue_key != new_finding.jira_issue.jira_key: - jira_helper.finding_unlink_jira(request, new_finding) - jira_helper.finding_link_jira( + jira_services.unlink_finding(request, new_finding) + jira_services.link_finding( request, new_finding, new_jira_issue_key, ) else: @@ -2178,7 +2178,7 @@ def promote_to_finding(request, fid): if new_jira_issue_key: logger.debug( "finding has no jira issue yet, but jira issue specified in request. trying to link.") - jira_helper.finding_link_jira( + jira_services.link_finding( request, new_finding, new_jira_issue_key, ) @@ -2231,8 +2231,8 @@ def promote_to_finding(request, fid): if use_jira: jform = JIRAFindingForm( prefix="jiraform", - push_all=jira_helper.is_push_all_issues(test), - jira_project=jira_helper.get_jira_project(test), + push_all=jira_services.is_push_all_issues(test), + jira_project=jira_services.get_project(test), ) return render( @@ -3066,8 +3066,8 @@ def _bulk_push_to_jira(finds, form, note): for finding in finds if finding.has_finding_group and ( - jira_helper.is_push_all_issues(finding) - or jira_helper.is_keep_in_sync_with_jira(finding) + jira_services.is_push_all_issues(finding) + or jira_services.is_keep_in_sync(finding) or form.cleaned_data.get("push_to_jira") ) ) @@ -3075,22 +3075,22 @@ def _bulk_push_to_jira(finds, form, note): for group in finding_groups: if ( form.cleaned_data.get("push_to_jira") - or jira_helper.is_push_all_issues(group) - or jira_helper.is_keep_in_sync_with_jira(group) + or jira_services.is_push_all_issues(group) + or jira_services.is_keep_in_sync(group) ): ( can_be_pushed_to_jira, error_message, _error_code, - ) = jira_helper.can_be_pushed_to_jira(group) + ) = jira_services.can_be_pushed(group) if not can_be_pushed_to_jira: error_counts[error_message] += 1 - jira_helper.log_jira_cannot_be_pushed_reason(error_message, group) + jira_services.log_cannot_be_pushed_reason(error_message, group) else: logger.debug( "pushing to jira from finding.finding_bulk_update_all()", ) - jira_helper.push_to_jira(group) + jira_services.push(group) success_count += 1 for error_message, error_count in error_counts.items(): @@ -3110,40 +3110,40 @@ def _bulk_push_to_jira(finds, form, note): # not sure yet if we want to support bulk unlink, so leave as commented out for now # if form.cleaned_data['unlink_from_jira']: # if finding.has_jira_issue: - # jira_helper.finding_unlink_jira(request, finding) + # jira_services.unlink_finding(request, finding) # Because we never call finding.save() in a bulk update, we need to actually # push the JIRA stuff here, rather than in finding.save() # can't use helper as when push_all_jira_issues is True, # the checkbox gets disabled and is always false - # push_to_jira = jira_helper.is_push_to_jira(new_finding, + # push_to_jira = jira_services.is_push_to_jira(new_finding, # form.cleaned_data.get('push_to_jira')) if ( form.cleaned_data.get("push_to_jira") - or jira_helper.is_push_all_issues(finding) - or jira_helper.is_keep_in_sync_with_jira(finding) + or jira_services.is_push_all_issues(finding) + or jira_services.is_keep_in_sync(finding) ) and not finding.has_finding_group: ( can_be_pushed_to_jira, error_message, _error_code, - ) = jira_helper.can_be_pushed_to_jira(finding) + ) = jira_services.can_be_pushed(finding) if finding.has_jira_group_issue and not finding.has_jira_issue: error_message = ( "finding already pushed as part of Finding Group" ) error_counts[error_message] += 1 - jira_helper.log_jira_cannot_be_pushed_reason(error_message, finding) + jira_services.log_cannot_be_pushed_reason(error_message, finding) elif not can_be_pushed_to_jira: error_counts[error_message] += 1 - jira_helper.log_jira_cannot_be_pushed_reason(error_message, finding) + jira_services.log_cannot_be_pushed_reason(error_message, finding) else: logger.debug( "pushing to jira from finding.finding_bulk_update_all()", ) - jira_helper.push_to_jira(finding) + jira_services.push(finding) if note is not None and isinstance(note, Notes): - jira_helper.add_comment(finding, note) + jira_services.add_comment(finding, note) success_count += 1 for error_message, error_count in error_counts.items(): @@ -3500,7 +3500,7 @@ def unlink_jira(request, fid): ) if finding.has_jira_issue: try: - jira_helper.finding_unlink_jira(request, finding) + jira_services.unlink_finding(request, finding) messages.add_message( request, @@ -3543,7 +3543,7 @@ def push_to_jira(request, fid): # but cant't change too much now without having a test suite, # so leave as is for now with the addition warning message # to check alerts for background errors. - if jira_helper.push_to_jira(finding): + if jira_services.push(finding): messages.add_message( request, messages.SUCCESS, diff --git a/dojo/finding_group/views.py b/dojo/finding_group/views.py index 451d4dcd720..afdb897e98a 100644 --- a/dojo/finding_group/views.py +++ b/dojo/finding_group/views.py @@ -12,7 +12,6 @@ from django.views import View from django.views.decorators.http import require_POST -import dojo.jira_link.helper as jira_helper from dojo.authorization.authorization import user_has_permission_or_403 from dojo.authorization.authorization_decorators import user_is_authorized from dojo.authorization.roles_permissions import Permissions @@ -23,6 +22,7 @@ ) from dojo.finding.queries import prefetch_for_findings from dojo.forms import DeleteFindingGroupForm, EditFindingGroupForm, FindingBulkUpdateForm +from dojo.jira import services as jira_services from dojo.models import Engagement, Finding, Finding_Group, GITHUB_PKey, Global_Role, Product from dojo.product.queries import get_authorized_products from dojo.utils import Product_Tab, add_breadcrumb, get_page_items, get_setting, get_system_setting, get_words_for_field @@ -48,7 +48,7 @@ def view_finding_group(request, fgid): product = get_object_or_404(Product, id=pid) user_has_permission_or_403(request.user, product, Permissions.Product_View) product_tab = Product_Tab(product, title="Findings", tab="findings") - jira_project = jira_helper.get_jira_project(product) + jira_project = jira_services.get_project(product) github_config = GITHUB_PKey.objects.filter(product=pid).first() findings_filter = finding_filter_class(request.GET, findings, user=request.user, pid=pid) elif finding_group.test.engagement.id: @@ -56,7 +56,7 @@ def view_finding_group(request, fgid): engagement = get_object_or_404(Engagement, id=eid) user_has_permission_or_403(request.user, engagement, Permissions.Engagement_View) product_tab = Product_Tab(engagement.product, title=engagement.name, tab="engagements") - jira_project = jira_helper.get_jira_project(engagement) + jira_project = jira_services.get_project(engagement) github_config = GITHUB_PKey.objects.filter(product__engagement=eid).first() findings_filter = finding_filter_class(request.GET, findings, user=request.user, eid=eid) @@ -82,7 +82,7 @@ def view_finding_group(request, fgid): if jira_issue: # See if the submitted issue was a issue key or the full URL - jira_project = jira_helper.get_jira_project(finding_group) + jira_project = jira_services.get_project(finding_group) if not jira_project or not jira_project.jira_instance: messages.add_message( request, @@ -94,13 +94,13 @@ def view_finding_group(request, fgid): jira_instance = jira_project.jira_instance jira_issue = jira_issue.removeprefix(jira_instance.url + "/browse/") - if finding_group.has_jira_issue and jira_issue != jira_helper.get_jira_key(finding_group): - jira_helper.unlink_jira(request, finding_group) - jira_helper.finding_group_link_jira(request, finding_group, jira_issue) + if finding_group.has_jira_issue and jira_issue != jira_services.get_key(finding_group): + jira_services.unlink(request, finding_group) + jira_services.link_finding_group(request, finding_group, jira_issue) elif not finding_group.has_jira_issue: - jira_helper.finding_group_link_jira(request, finding_group, jira_issue) + jira_services.link_finding_group(request, finding_group, jira_issue) elif push_to_jira: - jira_helper.push_to_jira(finding_group, sync=True) + jira_services.push(finding_group, sync=True) finding_group.save() return HttpResponseRedirect(reverse("view_test", args=(finding_group.test.id,))) @@ -162,7 +162,7 @@ def unlink_jira(request, fgid): logger.info("trying to unlink a linked jira issue from %d:%s", group.id, group.name) if group.has_jira_issue: try: - jira_helper.unlink_jira(request, group) + jira_services.unlink(request, group) messages.add_message( request, @@ -200,7 +200,7 @@ def push_to_jira(request, fgid): # it may look like success here, but the push_to_jira are swallowing exceptions # but cant't change too much now without having a test suite, so leave as is for now with the addition warning message to check alerts for background errors. - if jira_helper.push_to_jira(group, sync=True): + if jira_services.push(group, sync=True): messages.add_message( request, messages.SUCCESS, diff --git a/dojo/forms.py b/dojo/forms.py index f8c1b5f760a..211314915b3 100644 --- a/dojo/forms.py +++ b/dojo/forms.py @@ -1,5 +1,4 @@ import logging -import os import pickle import re import warnings @@ -16,13 +15,11 @@ from django.conf import settings from django.contrib.auth.models import Permission from django.contrib.auth.password_validation import validate_password -from django.core import validators from django.core.exceptions import ValidationError from django.core.validators import URLValidator from django.db.models import Count, Q from django.forms import modelformset_factory from django.forms.widgets import Select, Widget -from django.urls import reverse from django.utils import timezone from django.utils.dates import MONTHS from django.utils.safestring import mark_safe @@ -30,13 +27,26 @@ from polymorphic.base import ManagerInheritanceWarning from tagulous.forms import TagField -import dojo.jira_link.helper as jira_helper from dojo.authorization.authorization import user_has_configuration_permission, user_is_superuser_or_global_owner from dojo.authorization.roles_permissions import Permissions from dojo.endpoint.utils import endpoint_filter, endpoint_get_or_create, validate_endpoints_to_add from dojo.engagement.queries import get_authorized_engagements from dojo.finding.queries import get_authorized_findings from dojo.group.queries import get_authorized_groups, get_group_member_roles +from dojo.jira import services as jira_services +from dojo.jira.forms import ( # noqa: F401 backward compat + JIRA_TEMPLATE_CHOICES, + AdvancedJIRAForm, + BaseJiraForm, + DeleteJIRAInstanceForm, + JIRA_IssueForm, + JIRAEngagementForm, + JIRAFindingForm, + JIRAForm, + JIRAImportScanForm, + JIRAProjectForm, + get_jira_issue_template_dir_choices, +) from dojo.labels import get_labels from dojo.location.models import Location from dojo.location.utils import validate_locations_to_add @@ -73,9 +83,6 @@ GITHUB_Issue, GITHUB_PKey, Global_Role, - JIRA_Instance, - JIRA_Issue, - JIRA_Project, Note_Type, Notes, Notification_Webhooks, @@ -419,7 +426,7 @@ def __init__(self, *args, **kwargs): self.fields["push_to_jira"].label = "Push to JIRA" if hasattr(self.instance, "has_jira_issue") and self.instance.has_jira_issue: - jira_url = jira_helper.get_jira_url(self.instance) + jira_url = jira_services.get_url(self.instance) self.fields["jira_issue"].initial = jira_url self.fields["push_to_jira"].widget.attrs["checked"] = "checked" @@ -2830,89 +2837,6 @@ class Meta: "high_mapping_severity", "critical_mapping_severity", "finding_text"] -def get_jira_issue_template_dir_choices(): - template_root = settings.JIRA_TEMPLATE_ROOT - template_dir_list = [("", "---")] - for base_dir, dirnames, _filenames in os.walk(template_root): - # for filename in filenames: - # if base_dir.startswith(settings.TEMPLATE_DIR_PREFIX): - # base_dir = base_dir[len(settings.TEMPLATE_DIR_PREFIX):] - # template_list.append((os.path.join(base_dir, filename), filename)) - - for dirname in dirnames: - clean_base_dir = base_dir.removeprefix(settings.TEMPLATE_DIR_PREFIX) - template_dir_list.append((str(Path(clean_base_dir) / dirname), dirname)) - - logger.debug("templates: %s", template_dir_list) - return template_dir_list - - -JIRA_TEMPLATE_CHOICES = sorted(get_jira_issue_template_dir_choices()) - - -class JIRA_IssueForm(forms.ModelForm): - - class Meta: - model = JIRA_Issue - exclude = ["product"] - - -class BaseJiraForm(forms.ModelForm): - password = forms.CharField(widget=forms.PasswordInput, required=True, help_text=JIRA_Instance._meta.get_field("password").help_text, label=JIRA_Instance._meta.get_field("password").verbose_name) - - def test_jira_connection(self): - try: - # Attempt to validate the credentials before moving forward - jira_helper.get_jira_connection_raw(self.cleaned_data["url"], - self.cleaned_data["username"], - self.cleaned_data["password"]) - logger.debug("valid JIRA config!") - except Exception as e: - # form only used by admins, so we can show full error message using str(e) which can help debug any problems - message = "Unable to authenticate to JIRA. Please check the URL, username, password, captcha challenge, Network connection. Details in alert on top right. " + str( - e) - self.add_error("username", message) - self.add_error("password", message) - - def clean(self): - self.test_jira_connection() - return self.cleaned_data - - -class AdvancedJIRAForm(BaseJiraForm): - issue_template_dir = forms.ChoiceField(required=False, - choices=JIRA_TEMPLATE_CHOICES, - help_text="Choose the folder containing the Django templates used to render the JIRA issue description. These are stored in dojo/templates/issue-trackers. Leave empty to use the default jira_full templates.") - - def __init__(self, *args, **kwargs): - super().__init__(*args, **kwargs) - if self.instance: - self.fields["password"].required = False - - def clean(self): - if self.instance and not self.cleaned_data["password"]: - self.cleaned_data["password"] = self.instance.password - return super().clean() - - class Meta: - model = JIRA_Instance - exclude = [""] - - -class JIRAForm(BaseJiraForm): - issue_key = forms.CharField(required=True, help_text="A valid issue ID is required to gather the necessary information.") - issue_template_dir = forms.ChoiceField(required=False, - choices=JIRA_TEMPLATE_CHOICES, - help_text="Choose the folder containing the Django templates used to render the JIRA issue description. These are stored in dojo/templates/issue-trackers. Leave empty to use the default jira_full templates.") - - class Meta: - model = JIRA_Instance - exclude = ["product", "epic_name_id", "open_status_key", - "close_status_key", "info_mapping_severity", - "low_mapping_severity", "medium_mapping_severity", - "high_mapping_severity", "critical_mapping_severity", "finding_text"] - - class Benchmark_Product_SummaryForm(forms.ModelForm): class Meta: @@ -2953,15 +2877,6 @@ class Meta: fields = ["id"] -class DeleteJIRAInstanceForm(forms.ModelForm): - id = forms.IntegerField(required=True, - widget=forms.widgets.HiddenInput()) - - class Meta: - model = JIRA_Instance - fields = ["id"] - - class ToolTypeForm(forms.ModelForm): class Meta: model = Tool_Type @@ -3315,138 +3230,6 @@ class Meta: exclude = ["product"] -class JIRAProjectForm(forms.ModelForm): - inherit_from_product = forms.BooleanField(label="inherit JIRA settings from product", required=False) - jira_instance = forms.ModelChoiceField(queryset=JIRA_Instance.objects.all(), label="JIRA Instance", required=False) - issue_template_dir = forms.ChoiceField(required=False, - choices=JIRA_TEMPLATE_CHOICES, - help_text="Choose the folder containing the Django templates used to render the JIRA issue description. These are stored in dojo/templates/issue-trackers. Leave empty to use the default jira_full templates.") - - prefix = "jira-project-form" - - class Meta: - model = JIRA_Project - exclude = ["product", "engagement"] - fields = ["inherit_from_product", "jira_instance", "project_key", "issue_template_dir", "epic_issue_type_name", "component", "custom_fields", "jira_labels", "default_assignee", "enabled", "add_vulnerability_id_to_jira_label", "push_all_issues", "enable_engagement_epic_mapping", "push_notes", "product_jira_sla_notification", "risk_acceptance_expiration_notification"] - - def __init__(self, *args, **kwargs): - # if the form is shown for an engagement, we set a placeholder text around inherited settings from product - self.target = kwargs.pop("target", "product") - self.product = kwargs.pop("product", None) - self.engagement = kwargs.pop("engagement", None) - super().__init__(*args, **kwargs) - - logger.debug("self.target: %s, self.product: %s, self.instance: %s", self.target, self.product, self.instance) - logger.debug("data: %s", self.data) - if self.target == "engagement": - product_name = self.product.name if self.product else self.engagement.product.name if self.engagement.product else "" - - self.fields["project_key"].widget = forms.TextInput(attrs={"placeholder": f"JIRA settings inherited from product '{product_name}'"}) - self.fields["project_key"].help_text = f"JIRA settings are inherited from product '{product_name}', unless configured differently here." - self.fields["jira_instance"].help_text = f"JIRA settings are inherited from product '{product_name}' , unless configured differently here." - - # if we don't have an instance, django will insert a blank empty one :-( - # so we have to check for id to make sure we only trigger this when there is a real instance from db - if self.instance.id: - logger.debug("jira project instance found for engagement, unchecking inherit checkbox") - self.fields["jira_instance"].required = True - self.fields["project_key"].required = True - self.initial["inherit_from_product"] = False - # once a jira project config is attached to an engagement, we can't go back to inheriting - # because the config needs to remain in place for the existing jira issues - self.fields["inherit_from_product"].disabled = True - self.fields["inherit_from_product"].help_text = "Once an engagement has a JIRA Project stored, you cannot switch back to inheritance to avoid breaking existing JIRA issues" - self.fields["jira_instance"].disabled = False - self.fields["project_key"].disabled = False - self.fields["issue_template_dir"].disabled = False - self.fields["epic_issue_type_name"].disabled = False - self.fields["component"].disabled = False - self.fields["custom_fields"].disabled = False - self.fields["default_assignee"].disabled = False - self.fields["jira_labels"].disabled = False - self.fields["enabled"].disabled = False - self.fields["add_vulnerability_id_to_jira_label"].disabled = False - self.fields["push_all_issues"].disabled = False - self.fields["enable_engagement_epic_mapping"].disabled = False - self.fields["push_notes"].disabled = False - self.fields["product_jira_sla_notification"].disabled = False - self.fields["risk_acceptance_expiration_notification"].disabled = False - - elif self.product: - logger.debug("setting jira project fields from product1") - self.initial["inherit_from_product"] = True - jira_project_product = jira_helper.get_jira_project(self.product) - # we have to check that we are not in a POST request where jira project config data is posted - # this is because initial values will overwrite the actual values entered by the user - # makes no sense, but seems to be accepted behaviour: https://code.djangoproject.com/ticket/30407 - if jira_project_product and (self.prefix + "-jira_instance") not in self.data: - logger.debug("setting jira project fields from product2") - self.initial["jira_instance"] = jira_project_product.jira_instance.id if jira_project_product.jira_instance else None - self.initial["project_key"] = jira_project_product.project_key - self.initial["issue_template_dir"] = jira_project_product.issue_template_dir - self.initial["epic_issue_type_name"] = jira_project_product.epic_issue_type_name - self.initial["component"] = jira_project_product.component - self.initial["custom_fields"] = jira_project_product.custom_fields - self.initial["default_assignee"] = jira_project_product.default_assignee - self.initial["jira_labels"] = jira_project_product.jira_labels - self.initial["enabled"] = jira_project_product.enabled - self.initial["add_vulnerability_id_to_jira_label"] = jira_project_product.add_vulnerability_id_to_jira_label - self.initial["push_all_issues"] = jira_project_product.push_all_issues - self.initial["enable_engagement_epic_mapping"] = jira_project_product.enable_engagement_epic_mapping - self.initial["push_notes"] = jira_project_product.push_notes - self.initial["product_jira_sla_notification"] = jira_project_product.product_jira_sla_notification - self.initial["risk_acceptance_expiration_notification"] = jira_project_product.risk_acceptance_expiration_notification - - self.fields["jira_instance"].disabled = True - self.fields["project_key"].disabled = True - self.fields["issue_template_dir"].disabled = True - self.fields["epic_issue_type_name"].disabled = True - self.fields["component"].disabled = True - self.fields["custom_fields"].disabled = True - self.fields["default_assignee"].disabled = True - self.fields["jira_labels"].disabled = True - self.fields["enabled"].disabled = True - self.fields["add_vulnerability_id_to_jira_label"].disabled = True - self.fields["push_all_issues"].disabled = True - self.fields["enable_engagement_epic_mapping"].disabled = True - self.fields["push_notes"].disabled = True - self.fields["product_jira_sla_notification"].disabled = True - self.fields["risk_acceptance_expiration_notification"].disabled = True - - else: - del self.fields["inherit_from_product"] - - # if we don't have an instance, django will insert a blank empty one :-( - # so we have to check for id to make sure we only trigger this when there is a real instance from db - if self.instance.id: - self.fields["jira_instance"].required = True - self.fields["project_key"].required = True - self.fields["epic_issue_type_name"].required = True - - def clean(self): - logger.debug("validating jira project form") - cleaned_data = super().clean() - - logger.debug("clean: inherit: %s", self.cleaned_data.get("inherit_from_product", False)) - if not self.cleaned_data.get("inherit_from_product", False): - jira_instance = self.cleaned_data.get("jira_instance") - project_key = self.cleaned_data.get("project_key") - epic_issue_type_name = self.cleaned_data.get("epic_issue_type_name") - - if project_key and jira_instance and epic_issue_type_name: - return cleaned_data - - if not project_key and not jira_instance and not epic_issue_type_name: - return cleaned_data - - if self.target == "engagement": - msg = "JIRA Project needs a JIRA Instance, JIRA Project Key, and Epic issue type name, or choose to inherit settings from product" - raise ValidationError(msg) - msg = "JIRA Project needs a JIRA Instance, JIRA Project Key, and Epic issue type name, leave empty to have no JIRA integration setup" - raise ValidationError(msg) - return None - - class GITHUBFindingForm(forms.Form): def __init__(self, *args, **kwargs): self.enabled = kwargs.pop("enabled") @@ -3458,164 +3241,6 @@ def __init__(self, *args, **kwargs): push_to_github = forms.BooleanField(required=False) -class JIRAFindingForm(forms.Form): - def __init__(self, *args, **kwargs): - self.push_all = kwargs.pop("push_all", False) - self.instance = kwargs.pop("instance", None) - self.jira_project = kwargs.pop("jira_project", None) - # we provide the finding_form from the same page so we can add validation errors - # if the finding doesn't satisfy the rules to be pushed to JIRA - self.finding_form = kwargs.pop("finding_form", None) - - if self.instance is None and self.jira_project is None: - msg = "either and finding instance or jira_project is needed" - raise ValueError(msg) - - super().__init__(*args, **kwargs) - self.fields["push_to_jira"] = forms.BooleanField() - self.fields["push_to_jira"].required = False - if is_finding_groups_enabled(): - self.fields["push_to_jira"].help_text = "Checking this will overwrite content of your JIRA issue, or create one. If this finding is part of a Finding Group, the group will pushed instead of the finding." - else: - self.fields["push_to_jira"].help_text = "Checking this will overwrite content of your JIRA issue, or create one." - - self.fields["push_to_jira"].label = "Push to JIRA" - if self.push_all: - # This will show the checkbox as checked and greyed out, this way the user is aware - # that issues will be pushed to JIRA, given their product-level settings. - self.fields["push_to_jira"].help_text = ( - "Push all issues is enabled on this product. If you do not wish to push all issues" - " to JIRA, please disable Push all issues on this product." - ) - self.fields["push_to_jira"].widget.attrs["checked"] = "checked" - self.fields["push_to_jira"].disabled = True - - if self.instance: - if hasattr(self.instance, "has_jira_issue") and self.instance.has_jira_issue: - self.initial["jira_issue"] = self.instance.jira_issue.jira_key - self.fields["push_to_jira"].widget.attrs["checked"] = "checked" - if is_finding_groups_enabled(): - self.fields["jira_issue"].widget = forms.TextInput(attrs={"placeholder": "Leave empty and check push to jira to create a new JIRA issue for this finding, or the group this finding is in."}) - else: - self.fields["jira_issue"].widget = forms.TextInput(attrs={"placeholder": "Leave empty and check push to jira to create a new JIRA issue for this finding."}) - - if self.instance and hasattr(self.instance, "has_jira_group_issue") and self.instance.has_jira_group_issue: - self.fields["push_to_jira"].widget.attrs["checked"] = "checked" - self.fields["jira_issue"].help_text = "Changing the linked JIRA issue for finding groups is not (yet) supported." - self.initial["jira_issue"] = self.instance.finding_group.jira_issue.jira_key - self.fields["jira_issue"].disabled = True - - def clean(self): - logger.debug("jform clean") - super().clean() - jira_issue_key_new = self.cleaned_data.get("jira_issue") - finding = self.instance - jira_project = self.jira_project - - logger.debug("self.cleaned_data.push_to_jira: %s", self.cleaned_data.get("push_to_jira", None)) - - if self.cleaned_data.get("push_to_jira", None) and finding and finding.has_jira_group_issue: - can_be_pushed_to_jira, error_message, error_code = jira_helper.can_be_pushed_to_jira(finding.finding_group, self.finding_form) - if not can_be_pushed_to_jira: - self.add_error("push_to_jira", ValidationError(error_message, code=error_code)) - # for field in error_fields: - # self.finding_form.add_error(field, error) - - elif self.cleaned_data.get("push_to_jira", None) and finding: - can_be_pushed_to_jira, error_message, error_code = jira_helper.can_be_pushed_to_jira(finding, self.finding_form) - if not can_be_pushed_to_jira: - self.add_error("push_to_jira", ValidationError(error_message, code=error_code)) - # for field in error_fields: - # self.finding_form.add_error(field, error) - elif self.cleaned_data.get("push_to_jira", None): - active = self.finding_form["active"].value() - verified = self.finding_form["verified"].value() - if not active or (not verified and (get_system_setting("enforce_verified_status", True) or get_system_setting("enforce_verified_status_jira", True))): - logger.debug("Findings must be active and verified to be pushed to JIRA") - error_message = "Findings must be active and verified to be pushed to JIRA" - self.add_error("push_to_jira", ValidationError(error_message, code="not_active_or_verified")) - - if jira_issue_key_new and (not finding or not finding.has_jira_group_issue): - # when there is a group jira issue, we skip all the linking/unlinking as this is not supported (yet) - if finding: - # in theory there can multiple jira instances that have similar projects - # so checking by only the jira issue key can lead to false positives - # so we check also the jira internal id of the jira issue - # if the key and id are equal, it is probably the same jira instance and the same issue - # the database model is lacking some relations to also include the jira config name or url here - # and I don't want to change too much now. this should cover most usecases. - - jira_issue_need_to_exist = False - # changing jira link on finding - if finding.has_jira_issue and jira_issue_key_new != finding.jira_issue.jira_key: - jira_issue_need_to_exist = True - - # adding existing jira issue to finding without jira link - if not finding.has_jira_issue: - jira_issue_need_to_exist = True - - else: - jira_issue_need_to_exist = True - - if jira_issue_need_to_exist: - jira_issue_new = jira_helper.jira_get_issue(jira_project, jira_issue_key_new) - if not jira_issue_new: - raise ValidationError("JIRA issue " + jira_issue_key_new + " does not exist or cannot be retrieved") - - logger.debug("checking if provided jira issue id already is linked to another finding") - jira_issues = JIRA_Issue.objects.filter(jira_id=jira_issue_new.id, jira_key=jira_issue_key_new).exclude(engagement__isnull=False) - - if self.instance: - # just be sure we exclude the finding that is being edited - jira_issues = jira_issues.exclude(finding=finding) - - if len(jira_issues) > 0: - raise ValidationError("JIRA issue " + jira_issue_key_new + " already linked to " + reverse("view_finding", args=(jira_issues[0].finding_id,))) - - jira_issue = forms.CharField(required=False, label="Linked JIRA Issue", - validators=[validators.RegexValidator( - regex=r"^[A-Z][A-Z_0-9]+-\d+$", - message="JIRA issue key must be in XXXX-nnnn format ([A-Z][A-Z_0-9]+-\\d+)")]) - push_to_jira = forms.BooleanField(required=False, label="Push to JIRA") - - -class JIRAImportScanForm(forms.Form): - def __init__(self, *args, **kwargs): - self.push_all = kwargs.pop("push_all", False) - - super().__init__(*args, **kwargs) - if self.push_all: - # This will show the checkbox as checked and greyed out, this way the user is aware - # that issues will be pushed to JIRA, given their product-level settings. - self.fields["push_to_jira"].help_text = ( - "Push all issues is enabled on this product. If you do not wish to push all issues" - " to JIRA, please disable Push all issues on this product." - ) - self.fields["push_to_jira"].widget.attrs["checked"] = "checked" - self.fields["push_to_jira"].disabled = True - - push_to_jira = forms.BooleanField(required=False, label="Push to JIRA", help_text="Checking this will create a new jira issue for each new finding.") - - -class JIRAEngagementForm(forms.Form): - prefix = "jira-epic-form" - - def __init__(self, *args, **kwargs): - self.instance = kwargs.pop("instance", None) - - super().__init__(*args, **kwargs) - - if self.instance: - if self.instance.has_jira_issue: - self.fields["push_to_jira"].widget.attrs["checked"] = "checked" - self.fields["push_to_jira"].label = "Update JIRA Epic" - self.fields["push_to_jira"].help_text = "Checking this will update the existing EPIC in JIRA." - - push_to_jira = forms.BooleanField(required=False, label="Create EPIC", help_text="Checking this will create an EPIC in JIRA for this engagement.") - epic_name = forms.CharField(max_length=200, required=False, help_text="EPIC name in JIRA. If not specified, it defaults to the engagement name") - epic_priority = forms.CharField(max_length=200, required=False, help_text="EPIC priority. If not specified, the JIRA default priority will be used") - - class LoginBanner(forms.Form): banner_enable = forms.BooleanField( label="Enable login banner", diff --git a/dojo/importers/auto_create_context.py b/dojo/importers/auto_create_context.py index 26d37ae65b0..916bbea056e 100644 --- a/dojo/importers/auto_create_context.py +++ b/dojo/importers/auto_create_context.py @@ -181,6 +181,9 @@ def get_target_engagement_if_exists( """ if engagement := get_object_or_none(Engagement, pk=engagement_id): logger.debug("Using existing engagement by id: %s", engagement_id) + if product is not None and engagement.product_id != product.id: + msg = "The provided identifiers are inconsistent — the engagement does not belong to the specified product." + raise ValueError(msg) return engagement # if there's no product, then for sure there's no engagement either if product is None: @@ -203,6 +206,9 @@ def get_target_test_if_exists( """ if test := get_object_or_none(Test, pk=test_id): logger.debug("Using existing Test by id: %s", test_id) + if engagement is not None and test.engagement_id != engagement.id: + msg = "The provided identifiers are inconsistent — the test does not belong to the specified engagement." + raise ValueError(msg) return test # If the engagement is not supplied, we cannot do anything if not engagement: diff --git a/dojo/importers/base_importer.py b/dojo/importers/base_importer.py index c149f4e169d..85a1387d3aa 100644 --- a/dojo/importers/base_importer.py +++ b/dojo/importers/base_importer.py @@ -16,7 +16,7 @@ from dojo.celery_dispatch import dojo_dispatch_task from dojo.importers.location_manager import LocationManager, UnsavedLocation from dojo.importers.options import ImporterOptions -from dojo.jira_link.helper import is_keep_in_sync_with_jira +from dojo.jira.services import is_keep_in_sync from dojo.location.models import Location from dojo.models import ( # Import History States @@ -950,7 +950,7 @@ def mitigate_finding( # don't try to dedupe findings that we are closing finding.save(dedupe_option=False, product_grading_option=product_grading_option) else: - finding.save(dedupe_option=False, push_to_jira=(self.push_to_jira or is_keep_in_sync_with_jira(finding, prefetched_jira_instance=self.jira_instance)), product_grading_option=product_grading_option) + finding.save(dedupe_option=False, push_to_jira=(self.push_to_jira or is_keep_in_sync(finding, prefetched_jira_instance=self.jira_instance)), product_grading_option=product_grading_option) def notify_scan_added( self, diff --git a/dojo/importers/default_importer.py b/dojo/importers/default_importer.py index e8bd56baf55..6b3f3e1ec96 100644 --- a/dojo/importers/default_importer.py +++ b/dojo/importers/default_importer.py @@ -5,13 +5,12 @@ from django.db.models.query_utils import Q from django.urls import reverse -import dojo.jira_link.helper as jira_helper from dojo.celery_dispatch import dojo_dispatch_task from dojo.finding import helper as finding_helper from dojo.importers.base_importer import BaseImporter, Parser from dojo.importers.endpoint_manager import EndpointManager from dojo.importers.options import ImporterOptions -from dojo.jira_link.helper import is_keep_in_sync_with_jira +from dojo.jira import services as jira_services from dojo.models import ( Engagement, Finding, @@ -304,9 +303,9 @@ def process_findings( ) if self.push_to_jira: if findings[0].finding_group is not None: - jira_helper.push_to_jira(findings[0].finding_group) + jira_services.push(findings[0].finding_group) else: - jira_helper.push_to_jira(findings[0]) + jira_services.push(findings[0]) else: logger.debug("push_to_jira is False, not pushing to JIRA") @@ -413,8 +412,8 @@ def close_old_findings( if self.findings_groups_enabled and (self.push_to_jira or getattr(self.jira_instance, "finding_jira_sync", False)): for finding_group in {finding.finding_group for finding in old_findings if finding.finding_group is not None}: # Check the push_to_jira flag again to potentially shorty circuit without checking for existing findings - if self.push_to_jira or is_keep_in_sync_with_jira(finding_group, prefetched_jira_instance=self.jira_instance): - jira_helper.push_to_jira(finding_group) + if self.push_to_jira or jira_services.is_keep_in_sync(finding_group, prefetched_jira_instance=self.jira_instance): + jira_services.push(finding_group) # Calculate grade once after all findings have been closed if old_findings: diff --git a/dojo/importers/default_reimporter.py b/dojo/importers/default_reimporter.py index 2a22da10a35..d7759921240 100644 --- a/dojo/importers/default_reimporter.py +++ b/dojo/importers/default_reimporter.py @@ -6,7 +6,6 @@ from django.db.models.query_utils import Q import dojo.finding.helper as finding_helper -import dojo.jira_link.helper as jira_helper from dojo.celery_dispatch import dojo_dispatch_task from dojo.finding.deduplication import ( find_candidates_for_deduplication_hash, @@ -17,7 +16,7 @@ from dojo.importers.base_importer import BaseImporter, Parser from dojo.importers.endpoint_manager import EndpointManager from dojo.importers.options import ImporterOptions -from dojo.jira_link.helper import is_keep_in_sync_with_jira +from dojo.jira import services as jira_services from dojo.location.status import FindingLocationStatus from dojo.models import ( Development_Environment, @@ -564,8 +563,8 @@ def close_old_findings( if self.findings_groups_enabled and (self.push_to_jira or getattr(self.jira_instance, "finding_jira_sync", False)): for finding_group in {finding.finding_group for finding in findings if finding.finding_group is not None}: # Check the push_to_jira flag again to potentially shorty circuit without checking for existing findings - if self.push_to_jira or is_keep_in_sync_with_jira(finding_group, prefetched_jira_instance=self.jira_instance): - jira_helper.push_to_jira(finding_group) + if self.push_to_jira or jira_services.is_keep_in_sync(finding_group, prefetched_jira_instance=self.jira_instance): + jira_services.push(finding_group) # Calculate grade once after all findings have been closed if mitigated_findings: perform_product_grading(self.test.engagement.product) @@ -1065,8 +1064,8 @@ def process_groups_for_all_findings( if self.push_to_jira or getattr(self.jira_instance, "finding_jira_sync", False): object_to_push = findings[0].finding_group if findings[0].finding_group is not None else findings[0] # Check the push_to_jira flag again to potentially shorty circuit without checking for existing findings - if self.push_to_jira or is_keep_in_sync_with_jira(object_to_push, prefetched_jira_instance=self.jira_instance): - jira_helper.push_to_jira(object_to_push) + if self.push_to_jira or jira_services.is_keep_in_sync(object_to_push, prefetched_jira_instance=self.jira_instance): + jira_services.push(object_to_push) # We dont check if the finding jira sync is applicable quite yet until we can get in the loop # but this is a way to at least make it that far if self.findings_groups_enabled and (self.push_to_jira or getattr(self.jira_instance, "finding_jira_sync", False)): @@ -1076,8 +1075,8 @@ def process_groups_for_all_findings( if finding.finding_group is not None and not finding.is_mitigated }: # Check the push_to_jira flag again to potentially shorty circuit without checking for existing findings - if self.push_to_jira or is_keep_in_sync_with_jira(finding_group, prefetched_jira_instance=self.jira_instance): - jira_helper.push_to_jira(finding_group) + if self.push_to_jira or jira_services.is_keep_in_sync(finding_group, prefetched_jira_instance=self.jira_instance): + jira_services.push(finding_group) def calculate_unsaved_finding_hash_code( self, diff --git a/dojo/importers/options.py b/dojo/importers/options.py index 5b9576615c8..02cecff1113 100644 --- a/dojo/importers/options.py +++ b/dojo/importers/options.py @@ -10,7 +10,7 @@ from django.utils import timezone from django.utils.functional import SimpleLazyObject -from dojo.jira_link.helper import get_jira_instance +from dojo.jira.services import get_instance as get_jira_instance from dojo.models import ( Development_Environment, Dojo_User, diff --git a/dojo/jira_link/__init__.py b/dojo/jira/__init__.py similarity index 100% rename from dojo/jira_link/__init__.py rename to dojo/jira/__init__.py diff --git a/dojo/jira/api/__init__.py b/dojo/jira/api/__init__.py new file mode 100644 index 00000000000..e69de29bb2d diff --git a/dojo/jira/api/serializers.py b/dojo/jira/api/serializers.py new file mode 100644 index 00000000000..a4c6ae412d2 --- /dev/null +++ b/dojo/jira/api/serializers.py @@ -0,0 +1,91 @@ +import json + +from django.urls import reverse +from rest_framework import serializers + +from dojo.jira import services as jira_services +from dojo.models import ( + JIRA_Instance, + JIRA_Issue, + JIRA_Project, +) + + +class JIRAIssueSerializer(serializers.ModelSerializer): + url = serializers.SerializerMethodField(read_only=True) + + class Meta: + model = JIRA_Issue + fields = "__all__" + + def get_url(self, obj) -> str: + return jira_services.get_issue_url(obj) + + def validate(self, data): + if self.context["request"].method == "PATCH": + engagement = data.get("engagement", self.instance.engagement) + finding = data.get("finding", self.instance.finding) + finding_group = data.get( + "finding_group", self.instance.finding_group, + ) + else: + engagement = data.get("engagement", None) + finding = data.get("finding", None) + finding_group = data.get("finding_group", None) + + if ( + (engagement and not finding and not finding_group) + or (finding and not engagement and not finding_group) + or (finding_group and not engagement and not finding) + ): + pass + else: + msg = "Either engagement or finding or finding_group has to be set." + raise serializers.ValidationError(msg) + + if finding: + if (linked_finding := jira_services.already_linked(finding, data.get("jira_key"), data.get("jira_id"))) is not None: + msg = "JIRA issue " + data.get("jira_key") + " already linked to " + reverse("view_finding", args=(linked_finding.id,)) + raise serializers.ValidationError(msg) + + return data + + +class JIRAInstanceSerializer(serializers.ModelSerializer): + class Meta: + model = JIRA_Instance + fields = "__all__" + extra_kwargs = { + "password": {"write_only": True}, + } + + +class JIRAProjectSerializer(serializers.ModelSerializer): + class Meta: + model = JIRA_Project + fields = "__all__" + + def validate(self, data): + if self.context["request"].method == "PATCH": + engagement = data.get("engagement", self.instance.engagement) + product = data.get("product", self.instance.product) + else: + engagement = data.get("engagement", None) + product = data.get("product", None) + + if (engagement and product) or (not engagement and not product): + msg = "Either engagement or product has to be set." + raise serializers.ValidationError(msg) + + if "custom_fields" in data and isinstance(data["custom_fields"], str): + try: + data["custom_fields"] = json.loads(data["custom_fields"]) + except json.JSONDecodeError as e: + raise serializers.ValidationError({"custom_fields": f"Invalid JSON: {e}"}) from e + + return data + + +class EngagementUpdateJiraEpicSerializer(serializers.Serializer): + epic_name = serializers.CharField(required=False, max_length=200) + epic_priority = serializers.CharField(required=False, allow_null=True) diff --git a/dojo/jira/api/views.py b/dojo/jira/api/views.py new file mode 100644 index 00000000000..71fa5ee6393 --- /dev/null +++ b/dojo/jira/api/views.py @@ -0,0 +1,91 @@ +from django_filters.rest_framework import DjangoFilterBackend +from drf_spectacular.utils import extend_schema_view +from rest_framework.permissions import IsAuthenticated + +from dojo.api_v2 import permissions +from dojo.api_v2.views import DojoModelViewSet, PrefetchDojoModelViewSet, schema_with_prefetch +from dojo.authorization.roles_permissions import Permissions +from dojo.jira.api.serializers import ( + JIRAInstanceSerializer, + JIRAIssueSerializer, + JIRAProjectSerializer, +) +from dojo.jira.queries import ( + get_authorized_jira_issues, + get_authorized_jira_projects, +) +from dojo.models import ( + JIRA_Instance, + JIRA_Issue, + JIRA_Project, +) + + +class JiraInstanceViewSet( + DojoModelViewSet, +): + serializer_class = JIRAInstanceSerializer + queryset = JIRA_Instance.objects.none() + filter_backends = (DjangoFilterBackend,) + filterset_fields = ["id", "url"] + permission_classes = (permissions.UserHasConfigurationPermissionSuperuser,) + + def get_queryset(self): + return JIRA_Instance.objects.all().order_by("id") + + +# Authorization: object-based +# @extend_schema_view(**schema_with_prefetch()) +# Nested models with prefetch make the response schema too long for Swagger UI +class JiraIssuesViewSet( + PrefetchDojoModelViewSet, +): + serializer_class = JIRAIssueSerializer + queryset = JIRA_Issue.objects.none() + filter_backends = (DjangoFilterBackend,) + filterset_fields = [ + "id", + "jira_id", + "jira_key", + "finding", + "engagement", + "finding_group", + ] + + permission_classes = ( + IsAuthenticated, + permissions.UserHasJiraIssuePermission, + ) + + def get_queryset(self): + return get_authorized_jira_issues(Permissions.Product_View) + + +# Authorization: object-based +@extend_schema_view(**schema_with_prefetch()) +class JiraProjectViewSet( + PrefetchDojoModelViewSet, +): + serializer_class = JIRAProjectSerializer + queryset = JIRA_Project.objects.none() + filter_backends = (DjangoFilterBackend,) + filterset_fields = [ + "id", + "jira_instance", + "product", + "engagement", + "enabled", + "component", + "project_key", + "push_all_issues", + "enable_engagement_epic_mapping", + "push_notes", + ] + + permission_classes = ( + IsAuthenticated, + permissions.UserHasJiraProductPermission, + ) + + def get_queryset(self): + return get_authorized_jira_projects(Permissions.Product_View) diff --git a/dojo/jira/forms.py b/dojo/jira/forms.py new file mode 100644 index 00000000000..3ec7005d0fc --- /dev/null +++ b/dojo/jira/forms.py @@ -0,0 +1,404 @@ +import logging +import os +from pathlib import Path + +from django import forms +from django.conf import settings +from django.core import validators +from django.core.exceptions import ValidationError +from django.urls import reverse + +from dojo.jira import services as jira_services +from dojo.models import ( + JIRA_Instance, + JIRA_Issue, + JIRA_Project, +) +from dojo.utils import ( + get_system_setting, + is_finding_groups_enabled, +) + +logger = logging.getLogger(__name__) + + +def get_jira_issue_template_dir_choices(): + template_root = settings.JIRA_TEMPLATE_ROOT + template_dir_list = [("", "---")] + for base_dir, dirnames, _filenames in os.walk(template_root): + # for filename in filenames: + # if base_dir.startswith(settings.TEMPLATE_DIR_PREFIX): + # base_dir = base_dir[len(settings.TEMPLATE_DIR_PREFIX):] + # template_list.append((os.path.join(base_dir, filename), filename)) + + for dirname in dirnames: + clean_base_dir = base_dir.removeprefix(settings.TEMPLATE_DIR_PREFIX) + template_dir_list.append((str(Path(clean_base_dir) / dirname), dirname)) + + logger.debug("templates: %s", template_dir_list) + return template_dir_list + + +JIRA_TEMPLATE_CHOICES = sorted(get_jira_issue_template_dir_choices()) + + +class JIRA_IssueForm(forms.ModelForm): + + class Meta: + model = JIRA_Issue + exclude = ["product"] + + +class BaseJiraForm(forms.ModelForm): + password = forms.CharField(widget=forms.PasswordInput, required=True, help_text=JIRA_Instance._meta.get_field("password").help_text, label=JIRA_Instance._meta.get_field("password").verbose_name) + + def test_jira_connection(self): + try: + # Attempt to validate the credentials before moving forward + jira_services.get_connection_raw(self.cleaned_data["url"], + self.cleaned_data["username"], + self.cleaned_data["password"]) + logger.debug("valid JIRA config!") + except Exception as e: + # form only used by admins, so we can show full error message using str(e) which can help debug any problems + message = "Unable to authenticate to JIRA. Please check the URL, username, password, captcha challenge, Network connection. Details in alert on top right. " + str( + e) + self.add_error("username", message) + self.add_error("password", message) + + def clean(self): + self.test_jira_connection() + return self.cleaned_data + + +class AdvancedJIRAForm(BaseJiraForm): + issue_template_dir = forms.ChoiceField(required=False, + choices=JIRA_TEMPLATE_CHOICES, + help_text="Choose the folder containing the Django templates used to render the JIRA issue description. These are stored in dojo/templates/issue-trackers. Leave empty to use the default jira_full templates.") + + def __init__(self, *args, **kwargs): + super().__init__(*args, **kwargs) + if self.instance: + self.fields["password"].required = False + + def clean(self): + if self.instance and not self.cleaned_data["password"]: + self.cleaned_data["password"] = self.instance.password + return super().clean() + + class Meta: + model = JIRA_Instance + exclude = [""] + + +class JIRAForm(BaseJiraForm): + issue_key = forms.CharField(required=True, help_text="A valid issue ID is required to gather the necessary information.") + issue_template_dir = forms.ChoiceField(required=False, + choices=JIRA_TEMPLATE_CHOICES, + help_text="Choose the folder containing the Django templates used to render the JIRA issue description. These are stored in dojo/templates/issue-trackers. Leave empty to use the default jira_full templates.") + + class Meta: + model = JIRA_Instance + exclude = ["product", "epic_name_id", "open_status_key", + "close_status_key", "info_mapping_severity", + "low_mapping_severity", "medium_mapping_severity", + "high_mapping_severity", "critical_mapping_severity", "finding_text"] + + +class DeleteJIRAInstanceForm(forms.ModelForm): + id = forms.IntegerField(required=True, + widget=forms.widgets.HiddenInput()) + + class Meta: + model = JIRA_Instance + fields = ["id"] + + +class JIRAProjectForm(forms.ModelForm): + inherit_from_product = forms.BooleanField(label="inherit JIRA settings from product", required=False) + jira_instance = forms.ModelChoiceField(queryset=JIRA_Instance.objects.all(), label="JIRA Instance", required=False) + issue_template_dir = forms.ChoiceField(required=False, + choices=JIRA_TEMPLATE_CHOICES, + help_text="Choose the folder containing the Django templates used to render the JIRA issue description. These are stored in dojo/templates/issue-trackers. Leave empty to use the default jira_full templates.") + + prefix = "jira-project-form" + + class Meta: + model = JIRA_Project + exclude = ["product", "engagement"] + fields = ["inherit_from_product", "jira_instance", "project_key", "issue_template_dir", "epic_issue_type_name", "component", "custom_fields", "jira_labels", "default_assignee", "enabled", "add_vulnerability_id_to_jira_label", "push_all_issues", "enable_engagement_epic_mapping", "push_notes", "product_jira_sla_notification", "risk_acceptance_expiration_notification"] + + def __init__(self, *args, **kwargs): + # if the form is shown for an engagement, we set a placeholder text around inherited settings from product + self.target = kwargs.pop("target", "product") + self.product = kwargs.pop("product", None) + self.engagement = kwargs.pop("engagement", None) + super().__init__(*args, **kwargs) + + logger.debug("self.target: %s, self.product: %s, self.instance: %s", self.target, self.product, self.instance) + logger.debug("data: %s", self.data) + if self.target == "engagement": + product_name = self.product.name if self.product else self.engagement.product.name if self.engagement.product else "" + + self.fields["project_key"].widget = forms.TextInput(attrs={"placeholder": f"JIRA settings inherited from product '{product_name}'"}) + self.fields["project_key"].help_text = f"JIRA settings are inherited from product '{product_name}', unless configured differently here." + self.fields["jira_instance"].help_text = f"JIRA settings are inherited from product '{product_name}' , unless configured differently here." + + # if we don't have an instance, django will insert a blank empty one :-( + # so we have to check for id to make sure we only trigger this when there is a real instance from db + if self.instance.id: + logger.debug("jira project instance found for engagement, unchecking inherit checkbox") + self.fields["jira_instance"].required = True + self.fields["project_key"].required = True + self.initial["inherit_from_product"] = False + # once a jira project config is attached to an engagement, we can't go back to inheriting + # because the config needs to remain in place for the existing jira issues + self.fields["inherit_from_product"].disabled = True + self.fields["inherit_from_product"].help_text = "Once an engagement has a JIRA Project stored, you cannot switch back to inheritance to avoid breaking existing JIRA issues" + self.fields["jira_instance"].disabled = False + self.fields["project_key"].disabled = False + self.fields["issue_template_dir"].disabled = False + self.fields["epic_issue_type_name"].disabled = False + self.fields["component"].disabled = False + self.fields["custom_fields"].disabled = False + self.fields["default_assignee"].disabled = False + self.fields["jira_labels"].disabled = False + self.fields["enabled"].disabled = False + self.fields["add_vulnerability_id_to_jira_label"].disabled = False + self.fields["push_all_issues"].disabled = False + self.fields["enable_engagement_epic_mapping"].disabled = False + self.fields["push_notes"].disabled = False + self.fields["product_jira_sla_notification"].disabled = False + self.fields["risk_acceptance_expiration_notification"].disabled = False + + elif self.product: + logger.debug("setting jira project fields from product1") + self.initial["inherit_from_product"] = True + jira_project_product = jira_services.get_project(self.product) + # we have to check that we are not in a POST request where jira project config data is posted + # this is because initial values will overwrite the actual values entered by the user + # makes no sense, but seems to be accepted behaviour: https://code.djangoproject.com/ticket/30407 + if jira_project_product and (self.prefix + "-jira_instance") not in self.data: + logger.debug("setting jira project fields from product2") + self.initial["jira_instance"] = jira_project_product.jira_instance.id if jira_project_product.jira_instance else None + self.initial["project_key"] = jira_project_product.project_key + self.initial["issue_template_dir"] = jira_project_product.issue_template_dir + self.initial["epic_issue_type_name"] = jira_project_product.epic_issue_type_name + self.initial["component"] = jira_project_product.component + self.initial["custom_fields"] = jira_project_product.custom_fields + self.initial["default_assignee"] = jira_project_product.default_assignee + self.initial["jira_labels"] = jira_project_product.jira_labels + self.initial["enabled"] = jira_project_product.enabled + self.initial["add_vulnerability_id_to_jira_label"] = jira_project_product.add_vulnerability_id_to_jira_label + self.initial["push_all_issues"] = jira_project_product.push_all_issues + self.initial["enable_engagement_epic_mapping"] = jira_project_product.enable_engagement_epic_mapping + self.initial["push_notes"] = jira_project_product.push_notes + self.initial["product_jira_sla_notification"] = jira_project_product.product_jira_sla_notification + self.initial["risk_acceptance_expiration_notification"] = jira_project_product.risk_acceptance_expiration_notification + + self.fields["jira_instance"].disabled = True + self.fields["project_key"].disabled = True + self.fields["issue_template_dir"].disabled = True + self.fields["epic_issue_type_name"].disabled = True + self.fields["component"].disabled = True + self.fields["custom_fields"].disabled = True + self.fields["default_assignee"].disabled = True + self.fields["jira_labels"].disabled = True + self.fields["enabled"].disabled = True + self.fields["add_vulnerability_id_to_jira_label"].disabled = True + self.fields["push_all_issues"].disabled = True + self.fields["enable_engagement_epic_mapping"].disabled = True + self.fields["push_notes"].disabled = True + self.fields["product_jira_sla_notification"].disabled = True + self.fields["risk_acceptance_expiration_notification"].disabled = True + + else: + del self.fields["inherit_from_product"] + + # if we don't have an instance, django will insert a blank empty one :-( + # so we have to check for id to make sure we only trigger this when there is a real instance from db + if self.instance.id: + self.fields["jira_instance"].required = True + self.fields["project_key"].required = True + self.fields["epic_issue_type_name"].required = True + + def clean(self): + logger.debug("validating jira project form") + cleaned_data = super().clean() + + logger.debug("clean: inherit: %s", self.cleaned_data.get("inherit_from_product", False)) + if not self.cleaned_data.get("inherit_from_product", False): + jira_instance = self.cleaned_data.get("jira_instance") + project_key = self.cleaned_data.get("project_key") + epic_issue_type_name = self.cleaned_data.get("epic_issue_type_name") + + if project_key and jira_instance and epic_issue_type_name: + return cleaned_data + + if not project_key and not jira_instance and not epic_issue_type_name: + return cleaned_data + + if self.target == "engagement": + msg = "JIRA Project needs a JIRA Instance, JIRA Project Key, and Epic issue type name, or choose to inherit settings from product" + raise ValidationError(msg) + msg = "JIRA Project needs a JIRA Instance, JIRA Project Key, and Epic issue type name, leave empty to have no JIRA integration setup" + raise ValidationError(msg) + return None + + +class JIRAFindingForm(forms.Form): + def __init__(self, *args, **kwargs): + self.push_all = kwargs.pop("push_all", False) + self.instance = kwargs.pop("instance", None) + self.jira_project = kwargs.pop("jira_project", None) + # we provide the finding_form from the same page so we can add validation errors + # if the finding doesn't satisfy the rules to be pushed to JIRA + self.finding_form = kwargs.pop("finding_form", None) + + if self.instance is None and self.jira_project is None: + msg = "either and finding instance or jira_project is needed" + raise ValueError(msg) + + super().__init__(*args, **kwargs) + self.fields["push_to_jira"] = forms.BooleanField() + self.fields["push_to_jira"].required = False + if is_finding_groups_enabled(): + self.fields["push_to_jira"].help_text = "Checking this will overwrite content of your JIRA issue, or create one. If this finding is part of a Finding Group, the group will pushed instead of the finding." + else: + self.fields["push_to_jira"].help_text = "Checking this will overwrite content of your JIRA issue, or create one." + + self.fields["push_to_jira"].label = "Push to JIRA" + if self.push_all: + # This will show the checkbox as checked and greyed out, this way the user is aware + # that issues will be pushed to JIRA, given their product-level settings. + self.fields["push_to_jira"].help_text = ( + "Push all issues is enabled on this product. If you do not wish to push all issues" + " to JIRA, please disable Push all issues on this product." + ) + self.fields["push_to_jira"].widget.attrs["checked"] = "checked" + self.fields["push_to_jira"].disabled = True + + if self.instance: + if hasattr(self.instance, "has_jira_issue") and self.instance.has_jira_issue: + self.initial["jira_issue"] = self.instance.jira_issue.jira_key + self.fields["push_to_jira"].widget.attrs["checked"] = "checked" + if is_finding_groups_enabled(): + self.fields["jira_issue"].widget = forms.TextInput(attrs={"placeholder": "Leave empty and check push to jira to create a new JIRA issue for this finding, or the group this finding is in."}) + else: + self.fields["jira_issue"].widget = forms.TextInput(attrs={"placeholder": "Leave empty and check push to jira to create a new JIRA issue for this finding."}) + + if self.instance and hasattr(self.instance, "has_jira_group_issue") and self.instance.has_jira_group_issue: + self.fields["push_to_jira"].widget.attrs["checked"] = "checked" + self.fields["jira_issue"].help_text = "Changing the linked JIRA issue for finding groups is not (yet) supported." + self.initial["jira_issue"] = self.instance.finding_group.jira_issue.jira_key + self.fields["jira_issue"].disabled = True + + def clean(self): + logger.debug("jform clean") + super().clean() + jira_issue_key_new = self.cleaned_data.get("jira_issue") + finding = self.instance + jira_project = self.jira_project + + logger.debug("self.cleaned_data.push_to_jira: %s", self.cleaned_data.get("push_to_jira", None)) + + if self.cleaned_data.get("push_to_jira", None) and finding and finding.has_jira_group_issue: + can_be_pushed_to_jira, error_message, error_code = jira_services.can_be_pushed(finding.finding_group, self.finding_form) + if not can_be_pushed_to_jira: + self.add_error("push_to_jira", ValidationError(error_message, code=error_code)) + # for field in error_fields: + # self.finding_form.add_error(field, error) + + elif self.cleaned_data.get("push_to_jira", None) and finding: + can_be_pushed_to_jira, error_message, error_code = jira_services.can_be_pushed(finding, self.finding_form) + if not can_be_pushed_to_jira: + self.add_error("push_to_jira", ValidationError(error_message, code=error_code)) + # for field in error_fields: + # self.finding_form.add_error(field, error) + elif self.cleaned_data.get("push_to_jira", None): + active = self.finding_form["active"].value() + verified = self.finding_form["verified"].value() + if not active or (not verified and (get_system_setting("enforce_verified_status", True) or get_system_setting("enforce_verified_status_jira", True))): + logger.debug("Findings must be active and verified to be pushed to JIRA") + error_message = "Findings must be active and verified to be pushed to JIRA" + self.add_error("push_to_jira", ValidationError(error_message, code="not_active_or_verified")) + + if jira_issue_key_new and (not finding or not finding.has_jira_group_issue): + # when there is a group jira issue, we skip all the linking/unlinking as this is not supported (yet) + if finding: + # in theory there can multiple jira instances that have similar projects + # so checking by only the jira issue key can lead to false positives + # so we check also the jira internal id of the jira issue + # if the key and id are equal, it is probably the same jira instance and the same issue + # the database model is lacking some relations to also include the jira config name or url here + # and I don't want to change too much now. this should cover most usecases. + + jira_issue_need_to_exist = False + # changing jira link on finding + if finding.has_jira_issue and jira_issue_key_new != finding.jira_issue.jira_key: + jira_issue_need_to_exist = True + + # adding existing jira issue to finding without jira link + if not finding.has_jira_issue: + jira_issue_need_to_exist = True + + else: + jira_issue_need_to_exist = True + + if jira_issue_need_to_exist: + jira_issue_new = jira_services.jira_get_issue(jira_project, jira_issue_key_new) + if not jira_issue_new: + raise ValidationError("JIRA issue " + jira_issue_key_new + " does not exist or cannot be retrieved") + + logger.debug("checking if provided jira issue id already is linked to another finding") + jira_issues = JIRA_Issue.objects.filter(jira_id=jira_issue_new.id, jira_key=jira_issue_key_new).exclude(engagement__isnull=False) + + if self.instance: + # just be sure we exclude the finding that is being edited + jira_issues = jira_issues.exclude(finding=finding) + + if len(jira_issues) > 0: + raise ValidationError("JIRA issue " + jira_issue_key_new + " already linked to " + reverse("view_finding", args=(jira_issues[0].finding_id,))) + + jira_issue = forms.CharField(required=False, label="Linked JIRA Issue", + validators=[validators.RegexValidator( + regex=r"^[A-Z][A-Z_0-9]+-\d+$", + message="JIRA issue key must be in XXXX-nnnn format ([A-Z][A-Z_0-9]+-\\d+)")]) + push_to_jira = forms.BooleanField(required=False, label="Push to JIRA") + + +class JIRAImportScanForm(forms.Form): + def __init__(self, *args, **kwargs): + self.push_all = kwargs.pop("push_all", False) + + super().__init__(*args, **kwargs) + if self.push_all: + # This will show the checkbox as checked and greyed out, this way the user is aware + # that issues will be pushed to JIRA, given their product-level settings. + self.fields["push_to_jira"].help_text = ( + "Push all issues is enabled on this product. If you do not wish to push all issues" + " to JIRA, please disable Push all issues on this product." + ) + self.fields["push_to_jira"].widget.attrs["checked"] = "checked" + self.fields["push_to_jira"].disabled = True + + push_to_jira = forms.BooleanField(required=False, label="Push to JIRA", help_text="Checking this will create a new jira issue for each new finding.") + + +class JIRAEngagementForm(forms.Form): + prefix = "jira-epic-form" + + def __init__(self, *args, **kwargs): + self.instance = kwargs.pop("instance", None) + + super().__init__(*args, **kwargs) + + if self.instance: + if self.instance.has_jira_issue: + self.fields["push_to_jira"].widget.attrs["checked"] = "checked" + self.fields["push_to_jira"].label = "Update JIRA Epic" + self.fields["push_to_jira"].help_text = "Checking this will update the existing EPIC in JIRA." + + push_to_jira = forms.BooleanField(required=False, label="Create EPIC", help_text="Checking this will create an EPIC in JIRA for this engagement.") + epic_name = forms.CharField(max_length=200, required=False, help_text="EPIC name in JIRA. If not specified, it defaults to the engagement name") + epic_priority = forms.CharField(max_length=200, required=False, help_text="EPIC priority. If not specified, the JIRA default priority will be used") diff --git a/dojo/jira_link/helper.py b/dojo/jira/helper.py similarity index 100% rename from dojo/jira_link/helper.py rename to dojo/jira/helper.py diff --git a/dojo/jira/models.py b/dojo/jira/models.py new file mode 100644 index 00000000000..fe70c6ab3d0 --- /dev/null +++ b/dojo/jira/models.py @@ -0,0 +1,221 @@ +from django import forms +from django.conf import settings +from django.contrib import admin +from django.core.exceptions import ValidationError +from django.db import models +from django.utils.translation import gettext as _ + + +class JIRA_Instance(models.Model): + configuration_name = models.CharField(max_length=2000, help_text=_("Enter a name to give to this configuration"), default="") + url = models.URLField(max_length=2000, verbose_name=_("JIRA URL"), help_text=_("For more information how to configure Jira, read the DefectDojo documentation.")) + username = models.CharField(max_length=2000, verbose_name=_("Username/Email"), help_text=_("Username or Email Address, see DefectDojo documentation for more information.")) + password = models.CharField(max_length=2000, verbose_name=_("Password/Token"), help_text=_("Password or API Token, see DefectDojo documentation for more information.")) + + if hasattr(settings, "JIRA_ISSUE_TYPE_CHOICES_CONFIG"): + default_issue_type_choices = settings.JIRA_ISSUE_TYPE_CHOICES_CONFIG + else: + default_issue_type_choices = ( + ("Task", "Task"), + ("Story", "Story"), + ("Epic", "Epic"), + ("Spike", "Spike"), + ("Bug", "Bug"), + ("Security", "Security"), + ) + default_issue_type = models.CharField(max_length=255, + choices=default_issue_type_choices, + default="Bug", + help_text=_("You can define extra issue types in settings.py")) + issue_template_dir = models.CharField(max_length=255, + null=True, + blank=True, + help_text=_("Choose the folder containing the Django templates used to render the JIRA issue description. These are stored in dojo/templates/issue-trackers. Leave empty to use the default jira_full templates.")) + epic_name_id = models.IntegerField(help_text=_("To obtain the 'Epic name id' visit https://