From f1f68da25df487b19be97d4994123f4928b1f5d3 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Pedro=20Mart=C3=ADn?= Date: Mon, 11 Nov 2024 15:00:31 +0100 Subject: [PATCH] feat(jira): add jira integration (#5629) Co-authored-by: Pepe Fagoaga --- poetry.lock | 2 +- prowler/lib/outputs/jira/__init__.py | 0 .../lib/outputs/jira/exceptions/__init__.py | 0 .../lib/outputs/jira/exceptions/exceptions.py | 231 ++++ prowler/lib/outputs/jira/jira.py | 1166 +++++++++++++++++ tests/lib/outputs/jira/__init__.py | 0 tests/lib/outputs/jira/jira_test.py | 1041 +++++++++++++++ 7 files changed, 2439 insertions(+), 1 deletion(-) create mode 100644 prowler/lib/outputs/jira/__init__.py create mode 100644 prowler/lib/outputs/jira/exceptions/__init__.py create mode 100644 prowler/lib/outputs/jira/exceptions/exceptions.py create mode 100644 prowler/lib/outputs/jira/jira.py create mode 100644 tests/lib/outputs/jira/__init__.py create mode 100644 tests/lib/outputs/jira/jira_test.py diff --git a/poetry.lock b/poetry.lock index 020b9f7180..9c9eb0ccd2 100644 --- a/poetry.lock +++ b/poetry.lock @@ -1,4 +1,4 @@ -# This file is automatically @generated by Poetry 1.8.3 and should not be changed by hand. +# This file is automatically @generated by Poetry 1.7.1 and should not be changed by hand. [[package]] name = "about-time" diff --git a/prowler/lib/outputs/jira/__init__.py b/prowler/lib/outputs/jira/__init__.py new file mode 100644 index 0000000000..e69de29bb2 diff --git a/prowler/lib/outputs/jira/exceptions/__init__.py b/prowler/lib/outputs/jira/exceptions/__init__.py new file mode 100644 index 0000000000..e69de29bb2 diff --git a/prowler/lib/outputs/jira/exceptions/exceptions.py b/prowler/lib/outputs/jira/exceptions/exceptions.py new file mode 100644 index 0000000000..a8267ffba6 --- /dev/null +++ b/prowler/lib/outputs/jira/exceptions/exceptions.py @@ -0,0 +1,231 @@ +from prowler.exceptions.exceptions import ProwlerException + + +# Exceptions codes from 9000 to 9999 are reserved for Jira exceptions +class JiraBaseException(ProwlerException): + """Base class for Jira exceptions.""" + + JIRA_ERROR_CODES = { + (9000, "JiraNoProjectsError"): { + "message": "No projects were found in Jira.", + "remediation": "Please create a project in Jira.", + }, + (9001, "JiraAuthenticationError"): { + "message": "Failed to authenticate with Jira.", + "remediation": "Please check the connection settings and permissions and try again. Needed scopes are: read:jira-user read:jira-work write:jira-work", + }, + (9002, "JiraTestConnectionError"): { + "message": "Failed to connect to Jira.", + "remediation": "Please check the connection settings and permissions and try again.", + }, + (9003, "JiraCreateIssueError"): { + "message": "Failed to create an issue in Jira.", + "remediation": "Please check the connection settings and permissions and try again.", + }, + (9004, "JiraGetProjectsError"): { + "message": "Failed to get projects from Jira.", + "remediation": "Please check the connection settings and permissions and try again.", + }, + (9005, "JiraGetCloudIDError"): { + "message": "Failed to get the cloud ID from Jira.", + "remediation": "Please check the connection settings and permissions and try again.", + }, + (9006, "JiraGetCloudIDNoResourcesError"): { + "message": "No resources were found in Jira.", + "remediation": "Please check the connection settings and permissions and try again.", + }, + (9007, "JiraGetCloudIDResponseError"): { + "message": "Failed to get the cloud ID from Jira.", + "remediation": "Please check the connection settings and permissions and try again.", + }, + (9008, "JiraRefreshTokenResponseError"): { + "message": "Failed to refresh the access token, response code did not match 200.", + "remediation": "Please check the connection settings and permissions and try again.", + }, + (9009, "JiraRefreshTokenError"): { + "message": "Failed to refresh the access token.", + "remediation": "Please check the connection settings and permissions and try again.", + }, + (9010, "JiraGetAccessTokenError"): { + "message": "Failed to get the access token.", + "remediation": "Please check the connection settings and permissions and try again.", + }, + (9011, "JiraGetAuthResponseError"): { + "message": "Failed to authenticate with Jira.", + "remediation": "Please check the connection settings and permissions and try again.", + }, + (9012, "JiraGetProjectsResponseError"): { + "message": "Failed to get projects from Jira, response code did not match 200.", + "remediation": "Please check the connection settings and permissions and try again.", + }, + (9013, "JiraSendFindingsResponseError"): { + "message": "Failed to send findings to Jira, response code did not match 201.", + "remediation": "Please check the finding format and try again.", + }, + (9014, "JiraGetAvailableIssueTypesError"): { + "message": "Failed to get available issue types from Jira.", + "remediation": "Please check the connection settings and permissions and try again.", + }, + (9015, "JiraGetAvailableIssueTypesResponseError"): { + "message": "Failed to get available issue types from Jira, response code did not match 200.", + "remediation": "Please check the connection settings and permissions and try again.", + }, + (9016, "JiraInvalidIssueTypeError"): { + "message": "The issue type is invalid.", + "remediation": "Please check the issue type and try again.", + }, + (9017, "JiraNoTokenError"): { + "message": "No token was found.", + "remediation": "Make sure the token is set when using the Jira integration.", + }, + (9018, "JiraInvalidProjectKeyError"): { + "message": "The project key is invalid.", + "remediation": "Please check the project key and try again.", + }, + } + + def __init__(self, code, file=None, original_exception=None, message=None): + module = "Jira" + error_info = self.JIRA_ERROR_CODES.get((code, self.__class__.__name__)) + if message: + error_info["message"] = message + super().__init__( + code=code, + source=module, + file=file, + original_exception=original_exception, + error_info=error_info, + ) + + +class JiraNoProjectsError(JiraBaseException): + def __init__(self, file=None, original_exception=None, message=None): + super().__init__( + 9000, file=file, original_exception=original_exception, message=message + ) + + +class JiraAuthenticationError(JiraBaseException): + def __init__(self, file=None, original_exception=None, message=None): + super().__init__( + 9001, file=file, original_exception=original_exception, message=message + ) + + +class JiraTestConnectionError(JiraBaseException): + def __init__(self, file=None, original_exception=None, message=None): + super().__init__( + 9002, file=file, original_exception=original_exception, message=message + ) + + +class JiraCreateIssueError(JiraBaseException): + def __init__(self, file=None, original_exception=None, message=None): + super().__init__( + 9003, file=file, original_exception=original_exception, message=message + ) + + +class JiraGetProjectsError(JiraBaseException): + def __init__(self, file=None, original_exception=None, message=None): + super().__init__( + 9004, file=file, original_exception=original_exception, message=message + ) + + +class JiraGetCloudIDError(JiraBaseException): + def __init__(self, file=None, original_exception=None, message=None): + super().__init__( + 9005, file=file, original_exception=original_exception, message=message + ) + + +class JiraGetCloudIDNoResourcesError(JiraBaseException): + def __init__(self, file=None, original_exception=None, message=None): + super().__init__( + 9006, file=file, original_exception=original_exception, message=message + ) + + +class JiraGetCloudIDResponseError(JiraBaseException): + def __init__(self, file=None, original_exception=None, message=None): + super().__init__( + 9007, file=file, original_exception=original_exception, message=message + ) + + +class JiraRefreshTokenResponseError(JiraBaseException): + def __init__(self, file=None, original_exception=None, message=None): + super().__init__( + 9008, file=file, original_exception=original_exception, message=message + ) + + +class JiraRefreshTokenError(JiraBaseException): + def __init__(self, file=None, original_exception=None, message=None): + super().__init__( + 9009, file=file, original_exception=original_exception, message=message + ) + + +class JiraGetAccessTokenError(JiraBaseException): + def __init__(self, file=None, original_exception=None, message=None): + super().__init__( + 9010, file=file, original_exception=original_exception, message=message + ) + + +class JiraGetAuthResponseError(JiraBaseException): + def __init__(self, file=None, original_exception=None, message=None): + super().__init__( + 9011, file=file, original_exception=original_exception, message=message + ) + + +class JiraGetProjectsResponseError(JiraBaseException): + def __init__(self, file=None, original_exception=None, message=None): + super().__init__( + 9012, file=file, original_exception=original_exception, message=message + ) + + +class JiraSendFindingsResponseError(JiraBaseException): + def __init__(self, file=None, original_exception=None, message=None): + super().__init__( + 9013, file=file, original_exception=original_exception, message=message + ) + + +class JiraGetAvailableIssueTypesError(JiraBaseException): + def __init__(self, file=None, original_exception=None, message=None): + super().__init__( + 9014, file=file, original_exception=original_exception, message=message + ) + + +class JiraGetAvailableIssueTypesResponseError(JiraBaseException): + def __init__(self, file=None, original_exception=None, message=None): + super().__init__( + 9015, file=file, original_exception=original_exception, message=message + ) + + +class JiraInvalidIssueTypeError(JiraBaseException): + def __init__(self, file=None, original_exception=None, message=None): + super().__init__( + 9016, file=file, original_exception=original_exception, message=message + ) + + +class JiraNoTokenError(JiraBaseException): + def __init__(self, file=None, original_exception=None, message=None): + super().__init__( + 9017, file=file, original_exception=original_exception, message=message + ) + + +class JiraInvalidProjectKeyError(JiraBaseException): + def __init__(self, file=None, original_exception=None, message=None): + super().__init__( + 9018, file=file, original_exception=original_exception, message=message + ) diff --git a/prowler/lib/outputs/jira/jira.py b/prowler/lib/outputs/jira/jira.py new file mode 100644 index 0000000000..34629b7e42 --- /dev/null +++ b/prowler/lib/outputs/jira/jira.py @@ -0,0 +1,1166 @@ +import base64 +import os +from datetime import datetime, timedelta +from typing import Dict + +import requests +import requests.compat + +from prowler.lib.logger import logger +from prowler.lib.outputs.finding import Finding +from prowler.lib.outputs.jira.exceptions.exceptions import ( + JiraAuthenticationError, + JiraCreateIssueError, + JiraGetAccessTokenError, + JiraGetAuthResponseError, + JiraGetAvailableIssueTypesError, + JiraGetAvailableIssueTypesResponseError, + JiraGetCloudIDError, + JiraGetCloudIDNoResourcesError, + JiraGetCloudIDResponseError, + JiraGetProjectsError, + JiraGetProjectsResponseError, + JiraInvalidIssueTypeError, + JiraInvalidProjectKeyError, + JiraNoProjectsError, + JiraNoTokenError, + JiraRefreshTokenError, + JiraRefreshTokenResponseError, + JiraSendFindingsResponseError, + JiraTestConnectionError, +) +from prowler.providers.common.models import Connection + + +class Jira: + """ + Jira class to interact with the Jira API + + [Note] + This integration is limited to a single Jira Cloud, therefore all the issues will be created for same Jira Cloud ID. We will need to work on the ability of providing a Jira Cloud ID if the user is present in more than one. + + Attributes: + - _redirect_uri: The redirect URI + - _client_id: The client ID + - _client_secret: The client secret + - _access_token: The access token + - _refresh_token: The refresh token + - _expiration_date: The authentication expiration + - _cloud_id: The cloud ID + - _scopes: The scopes needed to authenticate, read:jira-user read:jira-work write:jira-work + - AUTH_URL: The URL to authenticate with Jira + - PARAMS_TEMPLATE: The template for the parameters to authenticate with Jira + - TOKEN_URL: The URL to get the access token from Jira + - API_TOKEN_URL: The URL to get the accessible resources from Jira + + Methods: + - __init__: Initialize the Jira object + - input_authorization_code: Input the authorization code + - auth_code_url: Generate the URL to authorize the application + - get_auth: Get the access token and refresh token + - get_cloud_id: Get the cloud ID from Jira + - get_access_token: Get the access token + - refresh_access_token: Refresh the access token from Jira + - test_connection: Test the connection to Jira and return a Connection object + - get_projects: Get the projects from Jira + - get_available_issue_types: Get the available issue types for a project + - send_findings: Send the findings to Jira and create an issue + + Raises: + - JiraGetAuthResponseError: Failed to get the access token and refresh token + - JiraGetCloudIDNoResourcesError: No resources were found in Jira when getting the cloud id + - JiraGetCloudIDResponseError: Failed to get the cloud ID, response code did not match 200 + - JiraGetCloudIDError: Failed to get the cloud ID from Jira + - JiraAuthenticationError: Failed to authenticate + - JiraRefreshTokenError: Failed to refresh the access token + - JiraRefreshTokenResponseError: Failed to refresh the access token, response code did not match 200 + - JiraGetAccessTokenError: Failed to get the access token + - JiraNoProjectsError: No projects found in Jira + - JiraGetProjectsError: Failed to get projects from Jira + - JiraGetProjectsResponseError: Failed to get projects from Jira, response code did not match 200 + - JiraInvalidIssueTypeError: The issue type is invalid + - JiraGetAvailableIssueTypesError: Failed to get available issue types from Jira + - JiraGetAvailableIssueTypesResponseError: Failed to get available issue types from Jira, response code did not match 200 + - JiraCreateIssueError: Failed to create an issue in Jira + - JiraSendFindingsResponseError: Failed to send the findings to Jira + - JiraTestConnectionError: Failed to test the connection + + Usage: + jira = Jira( + redirect_uri="http://localhost:8080", + client_id="client_id", + client_secret="client_secret + ) + jira.send_findings(findings=findings, project_key="KEY") + """ + + _redirect_uri: str = None + _client_id: str = None + _client_secret: str = None + _access_token: str = None + _refresh_token: str = None + _expiration_date: int = None + _cloud_id: str = None + _scopes: list[str] = None + AUTH_URL = "https://auth.atlassian.com/authorize" + PARAMS_TEMPLATE = { + "audience": "api.atlassian.com", + "client_id": None, + "scope": None, + "redirect_uri": None, + "state": None, + "response_type": "code", + "prompt": "consent", + } + TOKEN_URL = "https://auth.atlassian.com/oauth/token" + API_TOKEN_URL = "https://api.atlassian.com/oauth/token/accessible-resources" + + def __init__( + self, + redirect_uri: str = None, + client_id: str = None, + client_secret: str = None, + ): + self._redirect_uri = redirect_uri + self._client_id = client_id + self._client_secret = client_secret + self._scopes = ["read:jira-user", "read:jira-work", "write:jira-work"] + auth_url = self.auth_code_url() + authorization_code = self.input_authorization_code(auth_url) + self.get_auth(authorization_code) + + @property + def redirect_uri(self): + return self._redirect_uri + + @property + def client_id(self): + return self._client_id + + @property + def auth_expiration(self): + return self._expiration_date + + @auth_expiration.setter + def auth_expiration(self, value): + self._expiration_date = value + + @property + def cloud_id(self): + return self._cloud_id + + @cloud_id.setter + def cloud_id(self, value): + self._cloud_id = value + + @property + def scopes(self): + return self._scopes + + def get_params(self, state_encoded): + return { + **self.PARAMS_TEMPLATE, + "client_id": self.client_id, + "scope": " ".join(self.scopes), + "redirect_uri": self.redirect_uri, + "state": state_encoded, + } + + # TODO: Add static credentials for future use + @staticmethod + def input_authorization_code(auth_url: str = None) -> str: + """Input the authorization code + + Args: + - auth_url: The URL to authorize the application + + Returns: + - str: The authorization code from Jira + """ + print(f"Authorize the application by visiting this URL: {auth_url}") + return input("Enter the authorization code from Jira: ") + + def auth_code_url(self) -> str: + """Generate the URL to authorize the application + + Returns: + - str: The URL to authorize the application + + Raises: + - JiraGetAuthResponseError: Failed to get the access token and refresh token + """ + # Generate the state parameter + random_bytes = os.urandom(24) + state_encoded = base64.urlsafe_b64encode(random_bytes).decode("utf-8") + # Generate the URL + params = self.get_params(state_encoded) + + return f"{self.AUTH_URL}?{requests.compat.urlencode(params)}" + + @staticmethod + def get_timestamp_from_seconds(seconds: int) -> datetime: + """Get the timestamp adding the seconds to the current time + + Args: + - seconds: The seconds to add to the current time + + Returns: + - datetime: The timestamp with the seconds added + """ + return (datetime.now() + timedelta(seconds=seconds)).isoformat() + + def get_auth(self, auth_code: str = None) -> None: + """Get the access token and refresh token + + Args: + - auth_code: The authorization code from Jira + + Returns: + - None + + Raises: + - JiraGetAuthResponseError: Failed to get the access token and refresh token + - JiraGetCloudIDNoResourcesError: No resources were found in Jira when getting the cloud id + - JiraGetCloudIDResponseError: Failed to get the cloud ID, response code did not match 200 + - JiraGetCloudIDError: Failed to get the cloud ID from Jira + - JiraAuthenticationError: Failed to authenticate + - JiraRefreshTokenError: Failed to refresh the access token + - JiraRefreshTokenResponseError: Failed to refresh the access token, response code did not match 200 + - JiraGetAccessTokenError: Failed to get the access token + """ + try: + body = { + "grant_type": "authorization_code", + "client_id": self.client_id, + "client_secret": self._client_secret, + "code": auth_code, + "redirect_uri": self.redirect_uri, + } + + headers = {"Content-Type": "application/json"} + response = requests.post(self.TOKEN_URL, json=body, headers=headers) + + if response.status_code == 200: + tokens = response.json() + self._access_token = tokens.get("access_token") + self._refresh_token = tokens.get("refresh_token") + self._expiration_date = self.get_timestamp_from_seconds( + tokens.get("expires_in") + ) + self._cloud_id = self.get_cloud_id(self._access_token) + else: + response_error = ( + f"Failed to get auth: {response.status_code} - {response.json()}" + ) + raise JiraGetAuthResponseError( + message=response_error, file=os.path.basename(__file__) + ) + except JiraGetCloudIDNoResourcesError as no_resources_error: + raise no_resources_error + except JiraGetCloudIDResponseError as response_error: + raise response_error + except JiraGetCloudIDError as cloud_id_error: + raise cloud_id_error + except Exception as e: + message_error = f"Failed to get auth: {e}" + logger.error(message_error) + raise JiraAuthenticationError( + message=message_error, + file=os.path.basename(__file__), + ) + + def get_cloud_id(self, access_token: str = None) -> str: + """Get the cloud ID from Jira + + Args: + - access_token: The access token from Jira + + Returns: + - str: The cloud ID + + Raises: + - JiraGetCloudIDNoResourcesError: No resources were found in Jira when getting the cloud id + - JiraGetCloudIDResponseError: Failed to get the cloud ID, response code did not match 200 + - JiraGetCloudIDError: Failed to get the cloud ID from Jira + """ + try: + headers = {"Authorization": f"Bearer {access_token}"} + response = requests.get(self.API_TOKEN_URL, headers=headers) + + if response.status_code == 200: + resources = response.json() + if len(resources) > 0: + return resources[0].get("id") + else: + error_message = ( + "No resources were found in Jira when getting the cloud id" + ) + logger.error(error_message) + raise JiraGetCloudIDNoResourcesError( + message=error_message, + file=os.path.basename(__file__), + ) + else: + response_error = f"Failed to get cloud id: {response.status_code} - {response.json()}" + logger.warning(response_error) + raise JiraGetCloudIDResponseError( + message=response_error, file=os.path.basename(__file__) + ) + except Exception as e: + error_message = f"Failed to get the cloud ID from Jira: {e}" + logger.error(error_message) + raise JiraGetCloudIDError( + message=error_message, + file=os.path.basename(__file__), + ) + + def get_access_token(self) -> str: + """Get the access token + + Returns: + - str: The access token + + Raises: + - JiraRefreshTokenError: Failed to refresh the access token + - JiraRefreshTokenResponseError: Failed to refresh the access token, response code did not match 200 + - JiraGetAccessTokenError: Failed to get the access token + """ + try: + if self.auth_expiration and datetime.now() < datetime.fromisoformat( + self.auth_expiration + ): + return self._access_token + else: + return self.refresh_access_token() + except JiraRefreshTokenError as refresh_error: + raise refresh_error + except JiraRefreshTokenResponseError as response_error: + raise response_error + except Exception as e: + logger.error(f"Failed to get access token: {e}") + raise JiraGetAccessTokenError( + message="Failed to get the access token", + file=os.path.basename(__file__), + ) + + def refresh_access_token(self) -> str: + """Refresh the access token + + Returns: + - str: The access token + + Raises: + - JiraRefreshTokenError: Failed to refresh the access token + - JiraRefreshTokenResponseError: Failed to refresh the access token, response code did not match 200 + """ + try: + url = "https://auth.atlassian.com/oauth/token" + body = { + "grant_type": "refresh_token", + "client_id": self.client_id, + "client_secret": self._client_secret, + "refresh_token": self._refresh_token, + } + + headers = {"Content-Type": "application/json"} + response = requests.post(url, json=body, headers=headers) + + if response.status_code == 200: + tokens = response.json() + self._access_token = tokens.get("access_token") + self._refresh_token = tokens.get("refresh_token") + self._expiration_date = self.get_timestamp_from_seconds( + tokens.get("expires_in") + ) + return self._access_token + else: + response_error = f"Failed to refresh access token: {response.status_code} - {response.json()}" + logger.warning(response_error) + raise JiraRefreshTokenResponseError( + message=response_error, file=os.path.basename(__file__) + ) + + except Exception as e: + logger.error(f"Failed to refresh access token: {e}") + raise JiraRefreshTokenError( + message="Failed to refresh the access token", + file=os.path.basename(__file__), + ) + + @staticmethod + def test_connection( + redirect_uri: str = None, + client_id: str = None, + client_secret: str = None, + raise_on_exception: bool = True, + ) -> Connection: + """Test the connection to Jira + + Args: + - redirect_uri: The redirect URI + - client_id: The client ID + - client_secret: The client secret + - raise_on_exception: Whether to raise an exception or not + + Returns: + - Connection: The connection object + + Raises: + - JiraGetCloudIDNoResourcesError: No resources were found in Jira when getting the cloud id + - JiraGetCloudIDResponseError: Failed to get the cloud ID, response code did not match 200 + - JiraGetCloudIDError: Failed to get the cloud ID from Jira + - JiraAuthenticationError: Failed to authenticate + - JiraTestConnectionError: Failed to test the connection + """ + try: + jira = Jira( + redirect_uri=redirect_uri, + client_id=client_id, + client_secret=client_secret, + ) + access_token = jira.get_access_token() + + if not access_token: + return ValueError("Failed to get access token") + + headers = {"Authorization": f"Bearer {access_token}"} + response = requests.get( + f"https://api.atlassian.com/ex/jira/{jira.cloud_id}/rest/api/3/myself", + headers=headers, + ) + + if response.status_code == 200: + return Connection(is_connected=True) + else: + return Connection(is_connected=False, error=response.json()) + except JiraGetCloudIDNoResourcesError as no_resources_error: + logger.error( + f"{no_resources_error.__class__.__name__}[{no_resources_error.__traceback__.tb_lineno}]: {no_resources_error}" + ) + if raise_on_exception: + raise no_resources_error + return Connection(error=no_resources_error) + except JiraGetCloudIDResponseError as response_error: + logger.error( + f"{response_error.__class__.__name__}[{response_error.__traceback__.tb_lineno}]: {response_error}" + ) + if raise_on_exception: + raise response_error + return Connection(error=response_error) + except JiraGetCloudIDError as cloud_id_error: + logger.error( + f"{cloud_id_error.__class__.__name__}[{cloud_id_error.__traceback__.tb_lineno}]: {cloud_id_error}" + ) + if raise_on_exception: + raise cloud_id_error + return Connection(error=cloud_id_error) + except JiraAuthenticationError as auth_error: + logger.error( + f"{auth_error.__class__.__name__}[{auth_error.__traceback__.tb_lineno}]: {auth_error}" + ) + if raise_on_exception: + raise auth_error + return Connection(error=auth_error) + except Exception as error: + logger.error(f"Failed to test connection: {error}") + if raise_on_exception: + raise JiraTestConnectionError( + message="Failed to test connection on the Jira integration", + file=os.path.basename(__file__), + ) + return Connection(is_connected=False, error=error) + + def get_projects(self) -> Dict[str, str]: + """Get the projects from Jira + + Returns: + - list[Dict[str, str]]: The projects from Jira as a list of dictionaries, the projects format is [{"key": "KEY", "name": "NAME"}] + + Raises: + - JiraNoProjectsError: No projects found in Jira + - JiraGetProjectsError: Failed to get projects from Jira + - JiraRefreshTokenError: Failed to refresh the access token + - JiraRefreshTokenResponseError: Failed to refresh the access token, response code did not match + - JiraGetProjectsResponseError: Failed to get projects from Jira, response code did not match 200 + """ + try: + access_token = self.get_access_token() + + if not access_token: + return ValueError("Failed to get access token") + + headers = {"Authorization": f"Bearer {access_token}"} + response = requests.get( + f"https://api.atlassian.com/ex/jira/{self.cloud_id}/rest/api/3/project", + headers=headers, + ) + + if response.status_code == 200: + # Return the Project Key and Name, using only a dictionary + projects = { + project["key"]: project["name"] for project in response.json() + } + if len(projects) == 0: + logger.error("No projects found") + raise JiraNoProjectsError( + message="No projects found in Jira", + file=os.path.basename(__file__), + ) + return projects + else: + logger.error( + f"Failed to get projects: {response.status_code} - {response.json()}" + ) + raise JiraGetProjectsResponseError( + message="Failed to get projects from Jira", + file=os.path.basename(__file__), + ) + except JiraNoProjectsError as no_projects_error: + raise no_projects_error + except JiraRefreshTokenError as refresh_error: + raise refresh_error + except JiraRefreshTokenResponseError as response_error: + raise response_error + except Exception as e: + logger.error(f"Failed to get projects: {e}") + raise JiraGetProjectsError( + message="Failed to get projects from Jira", + file=os.path.basename(__file__), + ) + + def get_available_issue_types(self, project_key: str = None) -> list[str]: + """Get the available issue types for a project + + Args: + - project_key: The project key + + Returns: + - list[str]: The available issue types + + Raises: + - JiraRefreshTokenError: Failed to refresh the access token + - JiraRefreshTokenResponseError: Failed to refresh the access token, response code did not match 200 + - JiraGetAccessTokenError: Failed to get the access token + - JiraGetAuthResponseError: Failed to authenticate with Jira + - JiraGetProjectsError: Failed to get projects from Jira + - JiraGetProjectsResponseError: Failed to get projects from Jira, response code did not match 200 + """ + + try: + access_token = self.get_access_token() + + if not access_token: + return JiraNoTokenError( + message="No token was found", + file=os.path.basename(__file__), + ) + + headers = {"Authorization": f"Bearer {access_token}"} + response = requests.get( + f"https://api.atlassian.com/ex/jira/{self.cloud_id}/rest/api/3/issue/createmeta?projectKeys={project_key}&expand=projects.issuetypes.fields", + headers=headers, + ) + + if response.status_code == 200: + if len(response.json()["projects"]) == 0: + logger.error("No projects found") + raise JiraNoProjectsError( + message="No projects found in Jira", + file=os.path.basename(__file__), + ) + issue_types = response.json()["projects"][0]["issuetypes"] + return [issue_type["name"] for issue_type in issue_types] + else: + response_error = f"Failed to get available issue types: {response.status_code} - {response.json()}" + logger.warning(response_error) + raise JiraGetAvailableIssueTypesResponseError( + message=response_error, file=os.path.basename(__file__) + ) + except JiraRefreshTokenError as refresh_error: + raise refresh_error + except JiraRefreshTokenResponseError as response_error: + raise response_error + except Exception as e: + logger.error(f"Failed to get available issue types: {e}") + raise JiraGetAvailableIssueTypesError( + message="Failed to get available issue types", + file=os.path.basename(__file__), + ) + + @staticmethod + def get_color_from_status(status: str) -> str: + """Get the color from the status + + Args: + - status: The status of the finding + + Returns: + - str: The color of the status + """ + if status == "PASS": + return "#008000" + if status == "FAIL": + return "#FF0000" + if status == "MUTED": + return "#FFA500" + + @staticmethod + def get_adf_description( + check_id: str = None, + check_title: str = None, + severity: str = None, + status: str = None, + status_color: str = None, + status_extended: str = None, + provider: str = None, + region: str = None, + resource_uid: str = None, + resource_name: str = None, + risk: str = None, + recommendation_text: str = None, + recommendation_url: str = None, + ) -> dict: + return { + "type": "doc", + "version": 1, + "content": [ + { + "type": "paragraph", + "content": [ + { + "type": "text", + "text": "Prowler has discovered the following finding:", + } + ], + }, + { + "type": "table", + "attrs": {"layout": "full-width"}, + "content": [ + { + "type": "tableRow", + "content": [ + { + "type": "tableCell", + "attrs": {"colwidth": [1]}, + "content": [ + { + "type": "paragraph", + "content": [ + { + "type": "text", + "text": "Check Id", + "marks": [{"type": "strong"}], + } + ], + } + ], + }, + { + "type": "tableCell", + "attrs": {"colwidth": [3]}, + "content": [ + { + "type": "paragraph", + "content": [ + { + "type": "text", + "text": check_id, + "marks": [{"type": "code"}], + } + ], + } + ], + }, + ], + }, + { + "type": "tableRow", + "content": [ + { + "type": "tableCell", + "attrs": {"colwidth": [1]}, + "content": [ + { + "type": "paragraph", + "content": [ + { + "type": "text", + "text": "Check Title", + "marks": [{"type": "strong"}], + } + ], + } + ], + }, + { + "type": "tableCell", + "attrs": {"colwidth": [3]}, + "content": [ + { + "type": "paragraph", + "content": [ + { + "type": "text", + "text": check_title, + } + ], + } + ], + }, + ], + }, + { + "type": "tableRow", + "content": [ + { + "type": "tableCell", + "attrs": {"colwidth": [1]}, + "content": [ + { + "type": "paragraph", + "content": [ + { + "type": "text", + "text": "Severity", + "marks": [{"type": "strong"}], + } + ], + } + ], + }, + { + "type": "tableCell", + "attrs": {"colwidth": [3]}, + "content": [ + { + "type": "paragraph", + "content": [ + { + "type": "text", + "text": severity, + } + ], + } + ], + }, + ], + }, + { + "type": "tableRow", + "content": [ + { + "type": "tableCell", + "attrs": {"colwidth": [1]}, + "content": [ + { + "type": "paragraph", + "content": [ + { + "type": "text", + "text": "Status", + "marks": [{"type": "strong"}], + } + ], + } + ], + }, + { + "type": "tableCell", + "attrs": {"colwidth": [3]}, + "content": [ + { + "type": "paragraph", + "content": [ + { + "type": "text", + "text": status, + "marks": [ + {"type": "strong"}, + { + "type": "textColor", + "attrs": { + "color": status_color + }, + }, + ], + } + ], + } + ], + }, + ], + }, + { + "type": "tableRow", + "content": [ + { + "type": "tableCell", + "attrs": {"colwidth": [1]}, + "content": [ + { + "type": "paragraph", + "content": [ + { + "type": "text", + "text": "Status Extended", + "marks": [{"type": "strong"}], + } + ], + } + ], + }, + { + "type": "tableCell", + "attrs": {"colwidth": [3]}, + "content": [ + { + "type": "paragraph", + "content": [ + { + "type": "text", + "text": status_extended, + } + ], + } + ], + }, + ], + }, + { + "type": "tableRow", + "content": [ + { + "type": "tableCell", + "attrs": {"colwidth": [1]}, + "content": [ + { + "type": "paragraph", + "content": [ + { + "type": "text", + "text": "Provider", + "marks": [{"type": "strong"}], + } + ], + } + ], + }, + { + "type": "tableCell", + "attrs": {"colwidth": [3]}, + "content": [ + { + "type": "paragraph", + "content": [ + { + "type": "text", + "text": provider, + "marks": [{"type": "code"}], + } + ], + } + ], + }, + ], + }, + { + "type": "tableRow", + "content": [ + { + "type": "tableCell", + "attrs": {"colwidth": [1]}, + "content": [ + { + "type": "paragraph", + "content": [ + { + "type": "text", + "text": "Region", + "marks": [{"type": "strong"}], + } + ], + } + ], + }, + { + "type": "tableCell", + "attrs": {"colwidth": [3]}, + "content": [ + { + "type": "paragraph", + "content": [ + { + "type": "text", + "text": region, + "marks": [{"type": "code"}], + } + ], + } + ], + }, + ], + }, + { + "type": "tableRow", + "content": [ + { + "type": "tableCell", + "attrs": {"colwidth": [1]}, + "content": [ + { + "type": "paragraph", + "content": [ + { + "type": "text", + "text": "Resource UID", + "marks": [{"type": "strong"}], + } + ], + } + ], + }, + { + "type": "tableCell", + "attrs": {"colwidth": [3]}, + "content": [ + { + "type": "paragraph", + "content": [ + { + "type": "text", + "text": resource_uid, + "marks": [{"type": "code"}], + } + ], + } + ], + }, + ], + }, + { + "type": "tableRow", + "content": [ + { + "type": "tableCell", + "attrs": {"colwidth": [1]}, + "content": [ + { + "type": "paragraph", + "content": [ + { + "type": "text", + "text": "Resource Name", + "marks": [{"type": "strong"}], + } + ], + } + ], + }, + { + "type": "tableCell", + "attrs": {"colwidth": [3]}, + "content": [ + { + "type": "paragraph", + "content": [ + { + "type": "text", + "text": resource_name, + "marks": [{"type": "code"}], + } + ], + } + ], + }, + ], + }, + { + "type": "tableRow", + "content": [ + { + "type": "tableCell", + "attrs": {"colwidth": [1]}, + "content": [ + { + "type": "paragraph", + "content": [ + { + "type": "text", + "text": "Risk", + "marks": [{"type": "strong"}], + } + ], + } + ], + }, + { + "type": "tableCell", + "attrs": {"colwidth": [3]}, + "content": [ + { + "type": "paragraph", + "content": [ + { + "type": "text", + "text": risk, + } + ], + } + ], + }, + ], + }, + { + "type": "tableRow", + "content": [ + { + "type": "tableCell", + "attrs": {"colwidth": [1]}, + "content": [ + { + "type": "paragraph", + "content": [ + { + "type": "text", + "text": "Recommendation", + "marks": [{"type": "strong"}], + } + ], + } + ], + }, + { + "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 + }, + } + ], + }, + ], + } + ], + }, + ], + }, + ], + }, + ], + } + + def send_findings( + self, + findings: list[Finding] = None, + project_key: str = None, + issue_type: str = None, + ): + """ + Send the findings to Jira + + Args: + - findings: The findings to send + - project_key: The project key + - issue_type: The issue type + + Raises: + - JiraRefreshTokenError: Failed to refresh the access token + - JiraRefreshTokenResponseError: Failed to refresh the access token, response code did not match 200 + - JiraCreateIssueError: Failed to create an issue in Jira + - JiraSendFindingsResponseError: Failed to send the findings to Jira + """ + try: + access_token = self.get_access_token() + + if not access_token: + raise JiraNoTokenError( + message="No token was found", + file=os.path.basename(__file__), + ) + + projects = self.get_projects() + + if project_key not in projects: + logger.error("The project key is invalid") + raise JiraInvalidProjectKeyError( + message="The project key is invalid", + file=os.path.basename(__file__), + ) + + available_issue_types = self.get_available_issue_types(project_key) + + if issue_type not in available_issue_types: + logger.error("The issue type is invalid") + raise JiraInvalidIssueTypeError( + message="The issue type is invalid", file=os.path.basename(__file__) + ) + headers = { + "Authorization": f"Bearer {access_token}", + "Content-Type": "application/json", + } + + for finding in findings: + status_color = self.get_color_from_status(finding.status.value) + adf_description = self.get_adf_description( + check_id=finding.metadata.CheckID, + check_title=finding.metadata.CheckTitle, + severity=finding.metadata.Severity.value.upper(), + status=finding.status.value, + status_color=status_color, + status_extended=finding.status_extended, + provider=finding.metadata.Provider, + region=finding.region, + resource_uid=finding.resource_uid, + resource_name=finding.resource_name, + risk=finding.metadata.Risk, + recommendation_text=finding.metadata.Remediation.Recommendation.Text, + recommendation_url=finding.metadata.Remediation.Recommendation.Url, + ) + payload = { + "fields": { + "project": {"key": project_key}, + "summary": f"[Prowler] {finding.metadata.Severity.value.upper()} - {finding.metadata.CheckID} - {finding.resource_uid}", + "description": adf_description, + "issuetype": {"name": issue_type}, + } + } + + response = requests.post( + f"https://api.atlassian.com/ex/jira/{self.cloud_id}/rest/api/3/issue", + json=payload, + headers=headers, + ) + + if response.status_code != 201: + response_error = f"Failed to send finding: {response.status_code} - {response.json()}" + logger.warning(response_error) + raise JiraSendFindingsResponseError( + message=response_error, file=os.path.basename(__file__) + ) + else: + logger.info(f"Finding sent successfully: {response.json()}") + except JiraRefreshTokenError as refresh_error: + raise refresh_error + except JiraRefreshTokenResponseError as response_error: + raise response_error + except Exception as e: + logger.error(f"Failed to send findings: {e}") + raise JiraCreateIssueError( + message="Failed to create an issue in Jira", + file=os.path.basename(__file__), + ) diff --git a/tests/lib/outputs/jira/__init__.py b/tests/lib/outputs/jira/__init__.py new file mode 100644 index 0000000000..e69de29bb2 diff --git a/tests/lib/outputs/jira/jira_test.py b/tests/lib/outputs/jira/jira_test.py new file mode 100644 index 0000000000..238f433894 --- /dev/null +++ b/tests/lib/outputs/jira/jira_test.py @@ -0,0 +1,1041 @@ +from datetime import datetime, timedelta +from unittest.mock import MagicMock, PropertyMock, patch +from urllib.parse import parse_qs, urlparse + +import pytest +from freezegun import freeze_time + +from prowler.lib.outputs.jira.exceptions.exceptions import ( + JiraAuthenticationError, + JiraCreateIssueError, + JiraGetAvailableIssueTypesError, + JiraGetCloudIDError, + JiraGetProjectsError, + JiraNoProjectsError, + JiraRefreshTokenError, +) +from prowler.lib.outputs.jira.jira import Jira + +TEST_DATETIME = "2023-01-01T12:01:01+00:00" + + +class TestJiraIntegration: + @pytest.fixture(autouse=True) + @patch.object(Jira, "get_auth", return_value=None) + def setup(self, mock_get_auth, monkeypatch): + monkeypatch.setattr("builtins.input", lambda _: "test_authorization_code") + + # To disable vulture + mock_get_auth = mock_get_auth + + self.redirect_uri = "https://example.com/callback" + self.client_id = "test_client_id" + self.client_secret = "test_client_secret" + self.state_param = "unique_state_value" + + self.jira_integration = Jira( + redirect_uri=self.redirect_uri, + client_id=self.client_id, + client_secret=self.client_secret, + ) + + @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""" + + # To disable vulture + mock_get_auth = mock_get_auth + + generated_url = self.jira_integration.auth_code_url() + + expected_url = "https://auth.atlassian.com/authorize" + + parsed_url = urlparse(generated_url) + query_params = parse_qs(parsed_url.query) + + assert ( + parsed_url.scheme + "://" + parsed_url.netloc + parsed_url.path + == expected_url + ) + + assert query_params["audience"][0] == "api.atlassian.com" + assert query_params["client_id"][0] == self.client_id + assert ( + query_params["scope"][0] == "read:jira-user read:jira-work write:jira-work" + ) + assert query_params["redirect_uri"][0] == self.redirect_uri + assert query_params["state"][0] is not None + assert query_params["response_type"][0] == "code" + assert query_params["prompt"][0] == "consent" + + @freeze_time(TEST_DATETIME) + @patch("prowler.lib.outputs.jira.jira.requests.post") + @patch.object(Jira, "get_cloud_id", return_value="test_cloud_id") + def test_get_auth_successful(self, mock_get_cloud_id, mock_post): + """Test successful token retrieval in get_auth.""" + # To disable vulture + mock_get_cloud_id = mock_get_cloud_id + + mock_response = MagicMock() + mock_response.status_code = 200 + mock_response.json.return_value = { + "access_token": "test_access_token", + "refresh_token": "test_refresh_token", + "expires_in": 3600, + } + mock_post.return_value = mock_response + + self.jira_integration.get_auth("test_auth_code") + + assert self.jira_integration._access_token == "test_access_token" + assert self.jira_integration._refresh_token == "test_refresh_token" + assert ( + self.jira_integration._expiration_date + == (datetime.now() + timedelta(seconds=3600)).isoformat() + ) + assert self.jira_integration._cloud_id == "test_cloud_id" + + @patch( + "prowler.lib.outputs.jira.jira.requests.post", + side_effect=Exception("Connection error"), + ) + def test_get_auth_connection_error(self, mock_post): + """Test get_auth raises JiraAuthenticationError on connection failure.""" + # To disable vulture + mock_post = mock_post + + with pytest.raises(JiraAuthenticationError): + self.jira_integration.get_auth("test_auth_code") + + @patch("prowler.lib.outputs.jira.jira.requests.get") + def test_get_cloud_id_successful(self, mock_get): + """Test successful retrieval of cloud ID.""" + + mock_response = MagicMock() + mock_response.status_code = 200 + mock_response.json.return_value = [{"id": "test_cloud_id"}] + mock_get.return_value = mock_response + + cloud_id = self.jira_integration.get_cloud_id("test_access_token") + + assert cloud_id == "test_cloud_id" + + @patch("prowler.lib.outputs.jira.jira.requests.get") + def test_get_cloud_id_no_resources(self, mock_get): + """Test get_cloud_id raises JiraGetCloudIDNoResourcesError when no resources are found, later JiraGetCloudIDError will be raised.""" + + mock_response = MagicMock() + mock_response.status_code = 200 + mock_response.json.return_value = [] + mock_get.return_value = mock_response + + with pytest.raises(JiraGetCloudIDError): + self.jira_integration.get_cloud_id("test_access_token") + + @patch("prowler.lib.outputs.jira.jira.requests.get") + def test_get_cloud_id_response_error(self, mock_get): + """Test get_cloud_id raises JiraGetCloudIDResponseError when response code is not 200, later JiraGetCloudIDError will be raised.""" + + mock_response = MagicMock() + mock_response.status_code = 404 + mock_response.json.return_value = {"error": "Not Found"} + mock_get.return_value = mock_response + + with pytest.raises(JiraGetCloudIDError): + self.jira_integration.get_cloud_id("test_access_token") + + @patch( + "prowler.lib.outputs.jira.jira.requests.get", + side_effect=Exception("Connection error"), + ) + def test_get_cloud_id_unexpected_error(self, mock_get): + """Test get_cloud_id raises JiraGetCloudIDError on an unexpected exception.""" + # To disable vulture + mock_get = mock_get + + with pytest.raises(JiraGetCloudIDError): + self.jira_integration.get_cloud_id("test_access_token") + + @patch.object(Jira, "refresh_access_token", return_value="new_access_token") + def test_get_access_token_refresh(self, mock_refresh_access_token): + """Test get_access_token refreshes token when expired.""" + + self.jira_integration.auth_expiration = 0 + access_token = self.jira_integration.get_access_token() + + assert access_token == "new_access_token" + mock_refresh_access_token.assert_called_once() + + @freeze_time(TEST_DATETIME) + @patch("prowler.lib.outputs.jira.jira.requests.post") + @patch.object(Jira, "get_cloud_id", return_value="test_cloud_id") + def test_get_access_token_valid_token(self, mock_get_cloud_id, mock_post): + """Test successful token retrieval in get_auth.""" + # To disable vulture + mock_get_cloud_id = mock_get_cloud_id + + mock_response = MagicMock() + mock_response.status_code = 200 + mock_response.json.return_value = { + "access_token": "valid_access_token", + "refresh_token": "test_refresh_token", + "expires_in": 3600, + } + mock_post.return_value = mock_response + + self.jira_integration.get_auth("test_auth_code") + + access_token = self.jira_integration.get_access_token() + assert access_token == "valid_access_token" + + @patch.object(Jira, "refresh_access_token", side_effect=JiraRefreshTokenError) + def test_get_access_token_refresh_error(self, mock_refresh_access_token): + """Test get_access_token raises JiraRefreshTokenError on token refresh failure.""" + + # To disable vulture + mock_refresh_access_token = mock_refresh_access_token + + self.jira_integration.auth_expiration = ( + datetime.now() + timedelta(seconds=0) + ).isoformat() + + with pytest.raises(JiraRefreshTokenError): + self.jira_integration.get_access_token() + + @freeze_time(TEST_DATETIME) + @patch("prowler.lib.outputs.jira.jira.requests.post") + def test_refresh_access_token_successful(self, mock_post): + """Test successful access token refresh in refresh_access_token.""" + mock_response = MagicMock() + mock_response.status_code = 200 + expires_in_value = 3600 + mock_response.json.return_value = { + "access_token": "new_access_token", + "refresh_token": "new_refresh_token", + "expires_in": expires_in_value, + } + mock_post.return_value = mock_response + + new_access_token = self.jira_integration.refresh_access_token() + + assert new_access_token == "new_access_token" + assert self.jira_integration._access_token == "new_access_token" + assert self.jira_integration._refresh_token == "new_refresh_token" + assert ( + self.jira_integration._expiration_date + == (datetime.now() + timedelta(seconds=expires_in_value)).isoformat() + ) + + @patch("prowler.lib.outputs.jira.jira.requests.post") + def test_refresh_access_token_response_error(self, mock_post): + """Test refresh_access_token raises JiraRefreshTokenResponseError when response code is not 200, later JiraRefreshTokenError will be raised.""" + mock_response = MagicMock() + mock_response.status_code = 400 + mock_response.json.return_value = {"error": "invalid_request"} + mock_post.return_value = mock_response + + with pytest.raises(JiraRefreshTokenError): + self.jira_integration.refresh_access_token() + + @patch( + "prowler.lib.outputs.jira.jira.requests.post", + side_effect=Exception("Connection error"), + ) + def test_refresh_access_token_unexpected_error(self, mock_post): + """Test refresh_access_token raises JiraRefreshTokenError on unexpected exception.""" + # To disable vulture + mock_post = mock_post + + with pytest.raises(JiraRefreshTokenError): + self.jira_integration.refresh_access_token() + + @patch.object(Jira, "get_access_token", return_value="valid_access_token") + @patch.object(Jira, "get_cloud_id", return_value="test_cloud_id") + @patch.object(Jira, "get_auth", return_value=None) + @patch("prowler.lib.outputs.jira.jira.requests.get") + def test_test_connection_successful( + self, mock_get, mock_get_cloud_id, mock_get_auth, mock_get_access_token + ): + """Test that a successful connection returns an active Connection object.""" + # To disable vulture + mock_get_cloud_id = mock_get_cloud_id + mock_get_auth = mock_get_auth + mock_get_access_token = mock_get_access_token + + mock_response = MagicMock() + mock_response.status_code = 200 + mock_response.json.return_value = {"id": "test_user_id"} + mock_get.return_value = mock_response + + connection = self.jira_integration.test_connection( + redirect_uri=self.redirect_uri, + client_id=self.client_id, + client_secret=self.client_secret, + ) + + assert connection.is_connected + assert connection.error is None + + @patch.object( + Jira, + "get_access_token", + side_effect=JiraAuthenticationError("Failed to authenticate with Jira"), + ) + def test_test_connection_failed(self, mock_get_access_token): + """Test that a failed connection raises JiraAuthenticationError.""" + # To disable vulture + mock_get_access_token = mock_get_access_token + + with pytest.raises(JiraAuthenticationError): + self.jira_integration.test_connection( + redirect_uri=self.redirect_uri, + client_id=self.client_id, + client_secret=self.client_secret, + ) + + @patch.object(Jira, "get_access_token", return_value="valid_access_token") + @patch.object( + Jira, "cloud_id", new_callable=PropertyMock, return_value="test_cloud_id" + ) + @patch("prowler.lib.outputs.jira.jira.requests.get") + def test_get_projects_successful( + self, mock_get, mock_cloud_id, mock_get_access_token + ): + """Test successful retrieval of projects from Jira.""" + # To disable vulture + mock_cloud_id = mock_cloud_id + mock_get_access_token = mock_get_access_token + + mock_response = MagicMock() + mock_response.status_code = 200 + mock_response.json.return_value = [ + {"key": "PROJ1", "name": "Project One"}, + {"key": "PROJ2", "name": "Project Two"}, + ] + mock_get.return_value = mock_response + + projects = self.jira_integration.get_projects() + expected_projects = {"PROJ1": "Project One", "PROJ2": "Project Two"} + assert projects == expected_projects + + @patch.object(Jira, "get_access_token", return_value="valid_access_token") + @patch.object( + Jira, "cloud_id", new_callable=PropertyMock, return_value="test_cloud_id" + ) + @patch("prowler.lib.outputs.jira.jira.requests.get") + def test_get_projects_no_projects_found( + self, mock_get, mock_cloud_id, mock_get_access_token + ): + """Test that get_projects raises JiraNoProjectsError when no projects are found.""" + # To disable vulture + mock_cloud_id = mock_cloud_id + mock_get_access_token = mock_get_access_token + + mock_response = MagicMock() + mock_response.status_code = 200 + mock_response.json.return_value = [] + mock_get.return_value = mock_response + + with pytest.raises(JiraNoProjectsError): + self.jira_integration.get_projects() + + @patch.object(Jira, "get_access_token", return_value="valid_access_token") + @patch.object( + Jira, "cloud_id", new_callable=PropertyMock, return_value="test_cloud_id" + ) + @patch("prowler.lib.outputs.jira.jira.requests.get") + def test_get_projects_response_error( + self, mock_get, mock_cloud_id, mock_get_access_token + ): + """Test that get_projects raises JiraGetProjectsResponseError on non-200 response.""" + # To disable vulture + mock_cloud_id = mock_cloud_id + mock_get_access_token = mock_get_access_token + + mock_response = MagicMock() + mock_response.status_code = 404 + mock_response.json.return_value = {"error": "Not Found"} + mock_get.return_value = mock_response + + with pytest.raises(JiraGetProjectsError): + self.jira_integration.get_projects() + + @patch.object( + Jira, + "get_access_token", + side_effect=JiraRefreshTokenError("Failed to refresh the access token"), + ) + def test_get_projects_refresh_token_error(self, mock_get_access_token): + """Test that get_projects raises JiraRefreshTokenError when refreshing the token fails.""" + # To disable vulture + mock_get_access_token = mock_get_access_token + + with pytest.raises(JiraRefreshTokenError): + self.jira_integration.get_projects() + + @patch.object(Jira, "get_access_token", return_value="valid_access_token") + @patch.object( + Jira, "cloud_id", new_callable=PropertyMock, return_value="test_cloud_id" + ) + @patch("prowler.lib.outputs.jira.jira.requests.get") + def test_get_available_issue_types_successful( + self, mock_get, mock_cloud_id, mock_get_access_token + ): + """Test successful retrieval of issue types for a project.""" + # To disable vulture + mock_cloud_id = mock_cloud_id + mock_get_access_token = mock_get_access_token + + mock_response = MagicMock() + mock_response.status_code = 200 + mock_response.json.return_value = { + "projects": [ + { + "issuetypes": [ + {"name": "Bug"}, + {"name": "Task"}, + {"name": "Story"}, + ] + } + ] + } + mock_get.return_value = mock_response + + issue_types = self.jira_integration.get_available_issue_types( + project_key="TEST" + ) + + expected_issue_types = ["Bug", "Task", "Story"] + assert issue_types == expected_issue_types + + @patch.object(Jira, "get_access_token", return_value="valid_access_token") + @patch.object( + Jira, "cloud_id", new_callable=PropertyMock, return_value="test_cloud_id" + ) + @patch("prowler.lib.outputs.jira.jira.requests.get") + def test_get_available_issue_types_no_projects_found( + self, mock_get, mock_cloud_id, mock_get_access_token + ): + """Test that get_available_issue_types raises JiraNoProjectsError when no projects are found, later JiraGetAvailableIssueTypesError is raised.""" + # To disable vulture + mock_cloud_id = mock_cloud_id + mock_get_access_token = mock_get_access_token + + mock_response = MagicMock() + mock_response.status_code = 200 + mock_response.json.return_value = {"projects": []} + mock_get.return_value = mock_response + + with pytest.raises(JiraGetAvailableIssueTypesError): + self.jira_integration.get_available_issue_types(project_key="TEST") + + @patch.object(Jira, "get_access_token", return_value="valid_access_token") + @patch.object( + Jira, "cloud_id", new_callable=PropertyMock, return_value="test_cloud_id" + ) + @patch("prowler.lib.outputs.jira.jira.requests.get") + def test_get_available_issue_types_response_error( + self, mock_get, mock_cloud_id, mock_get_access_token + ): + """Test that get_available_issue_types raises JiraGetAvailableIssueTypesResponseError on non-200 response, JiraGetAvailableIssueTypesError will be raised later""" + # To disable vulture + mock_cloud_id = mock_cloud_id + mock_get_access_token = mock_get_access_token + + mock_response = MagicMock() + mock_response.status_code = 404 + mock_response.json.return_value = {"error": "Not Found"} + mock_get.return_value = mock_response + + with pytest.raises(JiraGetAvailableIssueTypesError): + self.jira_integration.get_available_issue_types(project_key="TEST") + + @patch.object( + Jira, + "get_access_token", + side_effect=JiraRefreshTokenError("Failed to refresh the access token"), + ) + def test_get_available_issue_types_refresh_token_error(self, mock_get_access_token): + """Test that get_available_issue_types raises JiraRefreshTokenError when refreshing the token fails.""" + # To disable vulture + mock_get_access_token = mock_get_access_token + + with pytest.raises(JiraRefreshTokenError): + self.jira_integration.get_available_issue_types(project_key="TEST") + + @patch.object(Jira, "get_access_token", return_value="valid_access_token") + @patch.object( + Jira, "get_available_issue_types", return_value=["Bug", "Task", "Story"] + ) + @patch.object(Jira, "get_projects", return_value={"TEST-1": "Test Project"}) + @patch("prowler.lib.outputs.jira.jira.requests.post") + def test_send_findings_successful( + self, + mock_post, + mock_get_available_issue_types, + mock_get_projects, + mock_get_access_token, + ): + """Test successful sending of findings to Jira.""" + # To disable vulture + mock_get_available_issue_types = mock_get_available_issue_types + mock_get_access_token = mock_get_access_token + mock_get_projects = mock_get_projects + + mock_response = MagicMock() + mock_response.status_code = 201 + mock_response.json.return_value = {"id": "12345", "key": "TEST-1"} + mock_post.return_value = mock_response + + finding = MagicMock() + finding.status.value = "FAIL" + finding.status_extended = "status_extended" + finding.metadata.Severity.value = "HIGH" + finding.metadata.CheckID = "CHECK-1" + finding.metadata.CheckTitle = "Check Title" + finding.resource_uid = "resource-1" + finding.resource_name = "resource_name" + finding.metadata.Provider = "aws" + finding.region = "region" + finding.metadata.Risk = "risk" + finding.metadata.Remediation.Recommendation.Text = "remediation_text" + finding.metadata.Remediation.Recommendation.Url = "remediation_url" + + self.jira_integration.cloud_id = "valid_cloud_id" + + self.jira_integration.send_findings( + findings=[finding], project_key="TEST-1", issue_type="Bug" + ) + + expected_url = ( + "https://api.atlassian.com/ex/jira/valid_cloud_id/rest/api/3/issue" + ) + expected_json = { + "fields": { + "project": {"key": "TEST-1"}, + "summary": "[Prowler] HIGH - CHECK-1 - resource-1", + "description": { + "type": "doc", + "version": 1, + "content": [ + { + "type": "paragraph", + "content": [ + { + "type": "text", + "text": "Prowler has discovered the following finding:", + } + ], + }, + { + "type": "table", + "attrs": {"layout": "full-width"}, + "content": [ + { + "type": "tableRow", + "content": [ + { + "type": "tableCell", + "attrs": {"colwidth": [1]}, + "content": [ + { + "type": "paragraph", + "content": [ + { + "type": "text", + "text": "Check Id", + "marks": [ + {"type": "strong"} + ], + } + ], + } + ], + }, + { + "type": "tableCell", + "attrs": {"colwidth": [3]}, + "content": [ + { + "type": "paragraph", + "content": [ + { + "type": "text", + "text": "CHECK-1", + "marks": [{"type": "code"}], + } + ], + } + ], + }, + ], + }, + { + "type": "tableRow", + "content": [ + { + "type": "tableCell", + "attrs": {"colwidth": [1]}, + "content": [ + { + "type": "paragraph", + "content": [ + { + "type": "text", + "text": "Check Title", + "marks": [ + {"type": "strong"} + ], + } + ], + } + ], + }, + { + "type": "tableCell", + "attrs": {"colwidth": [3]}, + "content": [ + { + "type": "paragraph", + "content": [ + { + "type": "text", + "text": "Check Title", + } + ], + } + ], + }, + ], + }, + { + "type": "tableRow", + "content": [ + { + "type": "tableCell", + "attrs": {"colwidth": [1]}, + "content": [ + { + "type": "paragraph", + "content": [ + { + "type": "text", + "text": "Severity", + "marks": [ + {"type": "strong"} + ], + } + ], + } + ], + }, + { + "type": "tableCell", + "attrs": {"colwidth": [3]}, + "content": [ + { + "type": "paragraph", + "content": [ + {"type": "text", "text": "HIGH"} + ], + } + ], + }, + ], + }, + { + "type": "tableRow", + "content": [ + { + "type": "tableCell", + "attrs": {"colwidth": [1]}, + "content": [ + { + "type": "paragraph", + "content": [ + { + "type": "text", + "text": "Status", + "marks": [ + {"type": "strong"} + ], + } + ], + } + ], + }, + { + "type": "tableCell", + "attrs": {"colwidth": [3]}, + "content": [ + { + "type": "paragraph", + "content": [ + { + "type": "text", + "text": "FAIL", + "marks": [ + {"type": "strong"}, + { + "type": "textColor", + "attrs": { + "color": "#FF0000" + }, + }, + ], + } + ], + } + ], + }, + ], + }, + { + "type": "tableRow", + "content": [ + { + "type": "tableCell", + "attrs": {"colwidth": [1]}, + "content": [ + { + "type": "paragraph", + "content": [ + { + "type": "text", + "text": "Status Extended", + "marks": [ + {"type": "strong"} + ], + } + ], + } + ], + }, + { + "type": "tableCell", + "attrs": {"colwidth": [3]}, + "content": [ + { + "type": "paragraph", + "content": [ + { + "type": "text", + "text": "status_extended", + } + ], + } + ], + }, + ], + }, + { + "type": "tableRow", + "content": [ + { + "type": "tableCell", + "attrs": {"colwidth": [1]}, + "content": [ + { + "type": "paragraph", + "content": [ + { + "type": "text", + "text": "Provider", + "marks": [ + {"type": "strong"} + ], + } + ], + } + ], + }, + { + "type": "tableCell", + "attrs": {"colwidth": [3]}, + "content": [ + { + "type": "paragraph", + "content": [ + { + "type": "text", + "text": "aws", + "marks": [{"type": "code"}], + } + ], + } + ], + }, + ], + }, + { + "type": "tableRow", + "content": [ + { + "type": "tableCell", + "attrs": {"colwidth": [1]}, + "content": [ + { + "type": "paragraph", + "content": [ + { + "type": "text", + "text": "Region", + "marks": [ + {"type": "strong"} + ], + } + ], + } + ], + }, + { + "type": "tableCell", + "attrs": {"colwidth": [3]}, + "content": [ + { + "type": "paragraph", + "content": [ + { + "type": "text", + "text": "region", + "marks": [{"type": "code"}], + } + ], + } + ], + }, + ], + }, + { + "type": "tableRow", + "content": [ + { + "type": "tableCell", + "attrs": {"colwidth": [1]}, + "content": [ + { + "type": "paragraph", + "content": [ + { + "type": "text", + "text": "Resource UID", + "marks": [ + {"type": "strong"} + ], + } + ], + } + ], + }, + { + "type": "tableCell", + "attrs": {"colwidth": [3]}, + "content": [ + { + "type": "paragraph", + "content": [ + { + "type": "text", + "text": "resource-1", + "marks": [{"type": "code"}], + } + ], + } + ], + }, + ], + }, + { + "type": "tableRow", + "content": [ + { + "type": "tableCell", + "attrs": {"colwidth": [1]}, + "content": [ + { + "type": "paragraph", + "content": [ + { + "type": "text", + "text": "Resource Name", + "marks": [ + {"type": "strong"} + ], + } + ], + } + ], + }, + { + "type": "tableCell", + "attrs": {"colwidth": [3]}, + "content": [ + { + "type": "paragraph", + "content": [ + { + "type": "text", + "text": "resource_name", + "marks": [{"type": "code"}], + } + ], + } + ], + }, + ], + }, + { + "type": "tableRow", + "content": [ + { + "type": "tableCell", + "attrs": {"colwidth": [1]}, + "content": [ + { + "type": "paragraph", + "content": [ + { + "type": "text", + "text": "Risk", + "marks": [ + {"type": "strong"} + ], + } + ], + } + ], + }, + { + "type": "tableCell", + "attrs": {"colwidth": [3]}, + "content": [ + { + "type": "paragraph", + "content": [ + {"type": "text", "text": "risk"} + ], + } + ], + }, + ], + }, + { + "type": "tableRow", + "content": [ + { + "type": "tableCell", + "attrs": {"colwidth": [1]}, + "content": [ + { + "type": "paragraph", + "content": [ + { + "type": "text", + "text": "Recommendation", + "marks": [ + {"type": "strong"} + ], + } + ], + } + ], + }, + { + "type": "tableCell", + "attrs": {"colwidth": [3]}, + "content": [ + { + "type": "paragraph", + "content": [ + { + "type": "text", + "text": "remediation_text ", + }, + { + "type": "text", + "text": "remediation_url", + "marks": [ + { + "type": "link", + "attrs": { + "href": "remediation_url" + }, + } + ], + }, + ], + } + ], + }, + ], + }, + ], + }, + ], + }, + "issuetype": {"name": "Bug"}, + } + } + expected_headers = { + "Authorization": "Bearer valid_access_token", + "Content-Type": "application/json", + } + + mock_post.assert_called_once_with( + expected_url, json=expected_json, headers=expected_headers + ) + + @patch.object(Jira, "get_access_token", return_value="valid_access_token") + @patch.object( + Jira, "get_available_issue_types", return_value=["Bug", "Task", "Story"] + ) + @patch("prowler.lib.outputs.jira.jira.requests.post") + def test_send_findings_invalid_issue_type( + self, mock_post, mock_get_available_issue_types, mock_get_access_token + ): + """Test that send_findings raises JiraInvalidIssueTypeError if the issue type is invalid that will raise JiraCreateIssueError later.""" + # To disable vulture + mock_get_available_issue_types = mock_get_available_issue_types + mock_get_access_token = mock_get_access_token + + with pytest.raises(JiraCreateIssueError): + self.jira_integration.send_findings( + findings=[MagicMock()], project_key="TEST", issue_type="InvalidType" + ) + + @patch.object(Jira, "get_access_token", return_value="valid_access_token") + @patch.object( + Jira, "get_available_issue_types", return_value=["Bug", "Task", "Story"] + ) + @patch("prowler.lib.outputs.jira.jira.requests.post") + def test_send_findings_response_error( + self, mock_post, mock_get_available_issue_types, mock_get_access_token + ): + """Test that send_findings raises JiraSendFindingsResponseError on non-201 response that will raise JiraCreateIssueError later.""" + # To disable vulture + mock_get_available_issue_types = mock_get_available_issue_types + mock_get_access_token = mock_get_access_token + + mock_response = MagicMock() + mock_response.status_code = 400 + mock_response.json.return_value = {"error": "Bad Request"} + mock_post.return_value = mock_response + + finding = MagicMock() + finding.status.value = "FAIL" + finding.metadata.Severity.value = "HIGH" + finding.metadata.CheckID = "CHECK-1" + finding.resource_uid = "resource-1" + + with pytest.raises(JiraCreateIssueError): + self.jira_integration.send_findings( + findings=[finding], project_key="TEST", issue_type="Bug" + ) + + @pytest.mark.parametrize( + "status, expected_color", + [("FAIL", "#FF0000"), ("PASS", "#008000"), ("MUTED", "#FFA500")], + ) + def test_get_color_from_status(self, status, expected_color): + """Test that get_color_from_status returns the correct color for a status.""" + assert self.jira_integration.get_color_from_status(status) == expected_color