diff --git a/prowler/CHANGELOG.md b/prowler/CHANGELOG.md index c5f899e656..2eaec5e7bd 100644 --- a/prowler/CHANGELOG.md +++ b/prowler/CHANGELOG.md @@ -13,6 +13,7 @@ All notable changes to the **Prowler SDK** are documented in this file. - Add explicit "name" field for each compliance framework and include "FRAMEWORK" and "NAME" in CSV output [(#7920)](https://github.com/prowler-cloud/prowler/pull/7920) - Equality validation for CheckID, filename and classname [(#8690)](https://github.com/prowler-cloud/prowler/pull/8690) - Improve logging for Security Hub integration [(#8608)](https://github.com/prowler-cloud/prowler/pull/8608) +- Support for Atlassian Document Format (ADF) in Jira integration [(#8878)](https://github.com/prowler-cloud/prowler/pull/8878) ### Changed diff --git a/prowler/lib/outputs/jira/jira.py b/prowler/lib/outputs/jira/jira.py index fa847d5fbc..6a80519b51 100644 --- a/prowler/lib/outputs/jira/jira.py +++ b/prowler/lib/outputs/jira/jira.py @@ -2,10 +2,12 @@ import base64 import os from dataclasses import dataclass from datetime import datetime, timedelta -from typing import Dict +from typing import Dict, List, Optional import requests import requests.compat +from markdown_it import MarkdownIt +from markdown_it.token import Token from prowler.lib.logger import logger from prowler.lib.outputs.finding import Finding @@ -47,6 +49,204 @@ class JiraConnection(Connection): projects: dict = None +class MarkdownToADFConverter: + """Helper to convert Markdown strings into Atlassian Document Format blocks.""" + + def __init__(self) -> None: + self._parser = MarkdownIt("commonmark", {"html": False}) + + def convert(self, text: Optional[str]) -> List[Dict]: + if text is None: + text = "" + + tokens = self._parser.parse(text) + if not tokens: + return [self._paragraph_with_text(text)] + + content_stack: List[List[Dict]] = [[]] + node_stack: List[Dict] = [] + + for token in tokens: + token_type = token.type + + if token_type == "paragraph_open": + node = {"type": "paragraph", "content": []} + node_stack.append(node) + content_stack.append(node["content"]) + elif token_type == "inline": + inline_nodes = self._convert_inline(token.children or []) + content_stack[-1].extend(inline_nodes) + elif token_type == "paragraph_close": + node = node_stack.pop() + content_stack.pop() + content_stack[-1].append(node) + elif token_type == "bullet_list_open": + node = {"type": "bulletList", "content": []} + node_stack.append(node) + content_stack.append(node["content"]) + elif token_type == "bullet_list_close": + node = node_stack.pop() + content_stack.pop() + content_stack[-1].append(node) + elif token_type == "ordered_list_open": + node: Dict = {"type": "orderedList", "content": []} + start_attr = token.attrGet("start") + if start_attr and start_attr.isdigit(): + start = int(start_attr) + if start != 1: + node["attrs"] = {"order": start} + node_stack.append(node) + content_stack.append(node["content"]) + elif token_type == "ordered_list_close": + node = node_stack.pop() + content_stack.pop() + content_stack[-1].append(node) + elif token_type == "list_item_open": + node = {"type": "listItem", "content": []} + node_stack.append(node) + content_stack.append(node["content"]) + elif token_type == "list_item_close": + node = node_stack.pop() + content_stack.pop() + content_stack[-1].append(node) + elif token_type == "heading_open": + level = self._safe_heading_level(token.tag) + node = {"type": "heading", "attrs": {"level": level}, "content": []} + node_stack.append(node) + content_stack.append(node["content"]) + elif token_type == "heading_close": + node = node_stack.pop() + content_stack.pop() + content_stack[-1].append(node) + elif token_type == "blockquote_open": + node = {"type": "blockquote", "content": []} + node_stack.append(node) + content_stack.append(node["content"]) + elif token_type == "blockquote_close": + node = node_stack.pop() + content_stack.pop() + content_stack[-1].append(node) + elif token_type in {"fence", "code_block"}: + language = None + if token_type == "fence": + info = (token.info or "").strip() + if info: + language = info.split()[0] + code_text = token.content.rstrip("\n") + code_node: Dict = { + "type": "codeBlock", + "content": [self._create_text_node(code_text, None)], + } + if language: + code_node["attrs"] = {"language": language} + content_stack[-1].append(code_node) + elif token_type in {"hr", "thematic_break"}: + content_stack[-1].append({"type": "rule"}) + elif token_type == "html_block": + html_text = token.content.strip() + if html_text: + content_stack[-1].append(self._paragraph_with_text(html_text)) + + result = content_stack[0] + if not result: + return [self._paragraph_with_text(text)] + + return result + + def _convert_inline(self, tokens: List[Token]) -> List[Dict]: + result: List[Dict] = [] + marks_stack: List[Dict] = [] + + for token in tokens: + token_type = token.type + + if token_type == "text": + result.extend(self._text_to_nodes(token.content, marks_stack)) + elif token_type == "code_inline": + marks = self._clone_marks(marks_stack) + marks.append({"type": "code"}) + result.append(self._create_text_node(token.content, marks)) + elif token_type in {"softbreak", "hardbreak"}: + result.append({"type": "hardBreak"}) + elif token_type == "strong_open": + marks_stack.append({"type": "strong"}) + elif token_type == "strong_close": + self._pop_mark(marks_stack, "strong") + elif token_type == "em_open": + marks_stack.append({"type": "em"}) + elif token_type == "em_close": + self._pop_mark(marks_stack, "em") + elif token_type == "link_open": + href = token.attrGet("href") or "" + mark: Dict = {"type": "link", "attrs": {"href": href}} + title = token.attrGet("title") + if title: + mark["attrs"]["title"] = title + marks_stack.append(mark) + elif token_type == "link_close": + self._pop_mark(marks_stack, "link") + elif token_type == "html_inline": + result.extend(self._text_to_nodes(token.content, marks_stack)) + elif token_type == "image": + alt_text = token.attrGet("alt") or token.content or "" + result.extend(self._text_to_nodes(alt_text, marks_stack)) + + return result + + @staticmethod + def _clone_marks(marks_stack: List[Dict]) -> List[Dict]: + cloned: List[Dict] = [] + for mark in marks_stack: + mark_copy = {"type": mark["type"]} + if "attrs" in mark: + mark_copy["attrs"] = dict(mark["attrs"]) + cloned.append(mark_copy) + return cloned + + def _text_to_nodes(self, text: str, marks_stack: List[Dict]) -> List[Dict]: + if not text: + return [] + + nodes: List[Dict] = [] + marks = self._clone_marks(marks_stack) + parts = text.split("\n") + + for index, part in enumerate(parts): + if part: + nodes.append(self._create_text_node(part, marks)) + if index < len(parts) - 1: + nodes.append({"type": "hardBreak"}) + + return nodes + + @staticmethod + def _create_text_node(text: str, marks: Optional[List[Dict]]) -> Dict: + node: Dict = {"type": "text", "text": text} + if marks: + node["marks"] = marks + return node + + def _paragraph_with_text(self, text: str) -> Dict: + return {"type": "paragraph", "content": [self._create_text_node(text, None)]} + + @staticmethod + def _pop_mark(marks_stack: List[Dict], mark_type: str) -> None: + for index in range(len(marks_stack) - 1, -1, -1): + if marks_stack[index]["type"] == mark_type: + marks_stack.pop(index) + break + + @staticmethod + def _safe_heading_level(tag: Optional[str]) -> int: + if tag and tag.startswith("h"): + try: + level = int(tag[1]) + return max(1, min(level, 6)) + except (ValueError, IndexError): + return 1 + return 1 + + class Jira: """ Jira class to interact with the Jira API @@ -112,6 +312,7 @@ class Jira: jira.send_findings(findings=findings, project_key="KEY") """ + _markdown_converter = MarkdownToADFConverter() _redirect_uri: str = None _client_id: str = None _client_secret: str = None @@ -173,6 +374,45 @@ class Jira: message=init_error, file=os.path.basename(__file__) ) + @staticmethod + def _build_code_block_content(code_value: str) -> Optional[Dict]: + if not code_value: + return None + + lines = code_value.splitlines() + if not lines: + return None + + language = None + first_line = lines[0].strip() + if first_line.startswith("```"): + language = first_line[3:].strip() or None + lines = lines[1:] + + while lines and not lines[0].strip(): + lines = lines[1:] + + if lines and lines[-1].strip().startswith("```"): + lines = lines[:-1] + + while lines and not lines[-1].strip(): + lines = lines[:-1] + + if not lines: + return None + + sanitized_text = "\n".join(lines) + + code_block: Dict = { + "type": "codeBlock", + "content": [{"type": "text", "text": sanitized_text}], + } + + if language: + code_block["attrs"] = {"language": language} + + return code_block + @property def redirect_uri(self): return self._redirect_uri @@ -837,8 +1077,8 @@ class Jira: return "#0000FF" return "#000000" # Default black color for unknown severities - @staticmethod def get_adf_description( + self, check_id: str = "", check_title: str = "", severity: str = "", @@ -1231,17 +1471,7 @@ class Jira: { "type": "tableCell", "attrs": {"colwidth": [3]}, - "content": [ - { - "type": "paragraph", - "content": [ - { - "type": "text", - "text": risk, - } - ], - } - ], + "content": self._markdown_converter.convert(risk), }, ], }, @@ -1340,6 +1570,34 @@ class Jira: ) # Add recommendation row + recommendation_content = self._markdown_converter.convert(recommendation_text) + if recommendation_url: + link_node = { + "type": "text", + "text": recommendation_url, + "marks": [{"type": "link", "attrs": {"href": recommendation_url}}], + } + + if ( + recommendation_content + and recommendation_content[-1].get("type") == "paragraph" + ): + paragraph = recommendation_content[-1] + paragraph_content = paragraph.setdefault("content", []) + if paragraph_content: + last_inline = paragraph_content[-1] + if last_inline.get("type") == "text" and not last_inline.get( + "text", "" + ).endswith(" "): + paragraph_content.append({"type": "text", "text": " "}) + elif last_inline.get("type") != "text": + paragraph_content.append({"type": "text", "text": " "}) + paragraph_content.append(link_node) + else: + recommendation_content.append( + {"type": "paragraph", "content": [link_node]} + ) + table_rows.append( { "type": "tableRow", @@ -1363,27 +1621,7 @@ class Jira: { "type": "tableCell", "attrs": {"colwidth": [3]}, - "content": [ - { - "type": "paragraph", - "content": [ - { - "type": "text", - "text": recommendation_text + " ", - }, - { - "type": "text", - "text": recommendation_url, - "marks": [ - { - "type": "link", - "attrs": {"href": recommendation_url}, - } - ], - }, - ], - } - ], + "content": recommendation_content, }, ], } @@ -1399,6 +1637,14 @@ class Jira: for code_type, code_value in remediation_codes: if code_value and code_value.strip(): + if code_type == "Other": + formatted_content = self._markdown_converter.convert(code_value) + else: + code_block = self._build_code_block_content(code_value) + if not code_block: + continue + formatted_content = [code_block] + table_rows.append( { "type": "tableRow", @@ -1422,18 +1668,7 @@ class Jira: { "type": "tableCell", "attrs": {"colwidth": [3]}, - "content": [ - { - "type": "paragraph", - "content": [ - { - "type": "text", - "text": code_value, - "marks": [{"type": "code"}], - } - ], - } - ], + "content": formatted_content, }, ], } @@ -1645,9 +1880,11 @@ class Jira: payload = { "fields": { "project": {"key": project_key}, - "summary": summary, + "summary": f"[Prowler] {finding.metadata.Severity.value.upper()} - {finding.metadata.CheckID} - {finding.resource_uid}", "description": adf_description, "issuetype": {"name": issue_type}, + "customfield_10148": {"value": "SDK"}, + "customfield_10088": {"value": "Core"}, } } if issue_labels: diff --git a/tests/lib/outputs/jira/jira_test.py b/tests/lib/outputs/jira/jira_test.py index 452b4a2a51..6e7f9f8220 100644 --- a/tests/lib/outputs/jira/jira_test.py +++ b/tests/lib/outputs/jira/jira_test.py @@ -1,5 +1,6 @@ import base64 from datetime import datetime, timedelta +from typing import List, Optional from unittest.mock import MagicMock, PropertyMock, patch from urllib.parse import parse_qs, urlparse @@ -60,6 +61,49 @@ class TestJiraIntegration: domain=self.domain, ) + @staticmethod + def _collect_text_from_cell(cell: dict) -> str: + pieces: List[str] = [] + + def walk(node: dict) -> None: + node_type = node.get("type") + if node_type == "text": + pieces.append(node.get("text", "")) + elif node_type == "hardBreak": + pieces.append(" ") + else: + for child in node.get("content", []): + walk(child) + if node_type in {"paragraph", "listItem"}: + pieces.append(" ") + + for child in cell.get("content", []): + walk(child) + + flattened = "".join(pieces) + return " ".join(flattened.split()) + + @staticmethod + def _find_link_mark(nodes: List[dict]) -> Optional[dict]: + for node in nodes: + if node.get("type") == "text": + for mark in node.get("marks", []): + if mark.get("type") == "link": + return mark + found = TestJiraIntegration._find_link_mark(node.get("content", [])) + if found: + return found + return None + + @staticmethod + def _find_table_row(rows: List[dict], header: str) -> dict: + for row in rows: + header_cell = row.get("content", [])[0] + header_text = TestJiraIntegration._collect_text_from_cell(header_cell) + if header_text == header: + return row + raise AssertionError(f"Row with header '{header}' not found") + @patch.object(Jira, "get_auth", return_value=None) def test_auth_code_url(self, mock_get_auth): """Test to verify the authorization URL generation with correct query parameters""" @@ -747,24 +791,24 @@ class TestJiraIntegration: assert table["type"] == "table" table_rows = table["content"] - row_texts = [] + row_entries = {} + recommendation_cell = None + for row in table_rows: - if row["type"] == "tableRow": - cells = row["content"] - if len(cells) == 2: - key_cell = cells[0]["content"][0]["content"][0]["text"] + if row["type"] != "tableRow": + continue - value_content = cells[1]["content"][0]["content"] - if len(value_content) > 1: - value_texts = [] - for content_item in value_content: - if content_item["type"] == "text": - value_texts.append(content_item["text"]) - value_cell = "".join(value_texts) - else: - value_cell = value_content[0]["text"] + cells = row["content"] + if len(cells) != 2: + continue - row_texts.append((key_cell, value_cell)) + key_text = self._collect_text_from_cell(cells[0]) + value_text = self._collect_text_from_cell(cells[1]) + + if key_text: + row_entries[key_text] = value_text + if key_text == "Recommendation": + recommendation_cell = cells[1] expected_keys = [ "Check Id", @@ -788,12 +832,15 @@ class TestJiraIntegration: "Tenant Info", ] - actual_keys = [key for key, _ in row_texts] - for expected_key in expected_keys: - assert expected_key in actual_keys, f"Missing row key: {expected_key}" + assert expected_key in row_entries, f"Missing row key: {expected_key}" - row_dict = dict(row_texts) + row_dict = row_entries + + assert recommendation_cell is not None + link_mark = self._find_link_mark(recommendation_cell.get("content", [])) + assert link_mark is not None + assert link_mark.get("attrs", {}).get("href") == "remediation_url" assert row_dict["Check Id"] == "CHECK-1" assert row_dict["Check Title"] == "Check Title" assert row_dict["Status"] == "FAIL" @@ -828,6 +875,117 @@ class TestJiraIntegration: assert "https://prowler-cloud-link/findings/12345" in row_dict["Finding URL"] assert "Tenant Info" in row_dict["Tenant Info"] + def test_get_adf_description_renders_markdown(self): + status_extended_md = "Finding uses **bold** text and `code` snippets." + risk_md = "High risk:\n- Item one\n- Item two" + recommendation_md = "Apply fixes:\n- Step one\n- Step two" + recommendation_url = "https://example.com/fix" + + adf_description = self.jira_integration.get_adf_description( + check_id="CHECK-1", + check_title="Sample check", + severity="HIGH", + severity_color="#FF0000", + status="FAIL", + status_color="#00FF00", + status_extended=status_extended_md, + provider="aws", + region="us-east-1", + resource_uid="resource-1", + resource_name="resource-name", + risk=risk_md, + recommendation_text=recommendation_md, + recommendation_url=recommendation_url, + ) + + assert adf_description["type"] == "doc" + table = adf_description["content"][1] + assert table["type"] == "table" + + rows = {} + for row in table["content"]: + if row.get("type") != "tableRow": + continue + key_cell, value_cell = row["content"] + key_text = self._collect_text_from_cell(key_cell) + rows[key_text] = value_cell + + assert "Status Extended" in rows + assert "Risk" in rows + assert "Recommendation" in rows + + def walk_nodes(nodes: List[dict]): + stack = list(nodes) + while stack: + current = stack.pop() + yield current + stack.extend(current.get("content", [])) + + status_text = self._collect_text_from_cell(rows["Status Extended"]) + assert status_text == status_extended_md + + risk_nodes = list(walk_nodes(rows["Risk"].get("content", []))) + assert any(node.get("type") == "bulletList" for node in risk_nodes) + + recommendation_cell = rows["Recommendation"] + recommendation_nodes = list(walk_nodes(recommendation_cell.get("content", []))) + assert any(node.get("type") == "bulletList" for node in recommendation_nodes) + link_mark = self._find_link_mark(recommendation_cell.get("content", [])) + assert link_mark is not None + assert link_mark.get("attrs", {}).get("href") == recommendation_url + + def test_get_adf_description_code_blocks_strip_fences(self): + code_block_value = """```hcl\nresource \"aws_s3_bucket\" \"example\" {\n bucket = \"my-bucket\"\n}\n```""" + + adf_description = self.jira_integration.get_adf_description( + check_id="CHECK-1", + check_title="Sample check", + severity="HIGH", + severity_color="#FF0000", + status="FAIL", + status_color="#00FF00", + recommendation_text="", + remediation_code_native_iac=code_block_value, + ) + + table = adf_description["content"][1] + code_row = self._find_table_row(table["content"], "Remediation Native IaC") + code_cell = code_row["content"][1] + code_block = code_cell["content"][0] + + assert code_block["type"] == "codeBlock" + assert code_block.get("attrs", {}).get("language") == "hcl" + expected_text = ( + 'resource "aws_s3_bucket" "example" {\n bucket = "my-bucket"\n}' + ) + assert code_block["content"][0]["text"] == expected_text + + def test_get_adf_description_other_remediation_uses_markdown(self): + other_value = "Use **bold** text" + + adf_description = self.jira_integration.get_adf_description( + check_id="CHECK-1", + check_title="Sample check", + severity="HIGH", + severity_color="#FF0000", + status="FAIL", + status_color="#00FF00", + recommendation_text="", + remediation_code_other=other_value, + ) + + table = adf_description["content"][1] + other_row = self._find_table_row(table["content"], "Remediation Other") + other_cell = other_row["content"][1] + + paragraph = other_cell["content"][0] + assert paragraph["type"] == "paragraph" + assert any( + mark.get("type") == "strong" + for node in paragraph.get("content", []) + for mark in node.get("marks", []) + ) + @patch.object(Jira, "get_access_token", return_value="valid_access_token") @patch.object( Jira, "get_available_issue_types", return_value=["Bug", "Task", "Story"]