feat(jira): support of ADF for MarkDown metadata fields (#8878)

This commit is contained in:
Hugo Pereira Brito
2025-10-09 12:31:31 +02:00
committed by GitHub
parent ecf749fce8
commit cc9aa7f7ee
3 changed files with 462 additions and 66 deletions
+1
View File
@@ -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
+284 -47
View File
@@ -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:
+177 -19
View File
@@ -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"]