From d460509262f696d843aacf4db362d531573cf79f Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Pedro=20Mart=C3=ADn?= Date: Mon, 17 Mar 2025 14:31:35 +0100 Subject: [PATCH] feat(jira): add basic auth method (#7233) --- .../lib/outputs/jira/exceptions/exceptions.py | 22 +++ prowler/lib/outputs/jira/jira.py | 129 ++++++++++++++++-- tests/lib/outputs/jira/jira_test.py | 87 ++++++++++++ 3 files changed, 224 insertions(+), 14 deletions(-) diff --git a/prowler/lib/outputs/jira/exceptions/exceptions.py b/prowler/lib/outputs/jira/exceptions/exceptions.py index a8267ffba6..ed138a73a4 100644 --- a/prowler/lib/outputs/jira/exceptions/exceptions.py +++ b/prowler/lib/outputs/jira/exceptions/exceptions.py @@ -82,6 +82,14 @@ class JiraBaseException(ProwlerException): "message": "The project key is invalid.", "remediation": "Please check the project key and try again.", }, + (9019, "JiraBasicAuthError"): { + "message": "Failed to authenticate with Jira using basic authentication.", + "remediation": "Please check the user mail and API token and try again.", + }, + (9020, "JiraInvalidParameterError"): { + "message": "Missing parameters on Jira Init function.", + "remediation": "Please check the parameters and try again.", + }, } def __init__(self, code, file=None, original_exception=None, message=None): @@ -229,3 +237,17 @@ class JiraInvalidProjectKeyError(JiraBaseException): super().__init__( 9018, file=file, original_exception=original_exception, message=message ) + + +class JiraBasicAuthError(JiraBaseException): + def __init__(self, file=None, original_exception=None, message=None): + super().__init__( + 9019, file=file, original_exception=original_exception, message=message + ) + + +class JiraInvalidParameterError(JiraBaseException): + def __init__(self, file=None, original_exception=None, message=None): + super().__init__( + 9020, file=file, original_exception=original_exception, message=message + ) diff --git a/prowler/lib/outputs/jira/jira.py b/prowler/lib/outputs/jira/jira.py index 34629b7e42..616f97b820 100644 --- a/prowler/lib/outputs/jira/jira.py +++ b/prowler/lib/outputs/jira/jira.py @@ -10,6 +10,7 @@ from prowler.lib.logger import logger from prowler.lib.outputs.finding import Finding from prowler.lib.outputs.jira.exceptions.exceptions import ( JiraAuthenticationError, + JiraBasicAuthError, JiraCreateIssueError, JiraGetAccessTokenError, JiraGetAuthResponseError, @@ -21,6 +22,7 @@ from prowler.lib.outputs.jira.exceptions.exceptions import ( JiraGetProjectsError, JiraGetProjectsResponseError, JiraInvalidIssueTypeError, + JiraInvalidParameterError, JiraInvalidProjectKeyError, JiraNoProjectsError, JiraNoTokenError, @@ -84,6 +86,8 @@ class Jira: - JiraCreateIssueError: Failed to create an issue in Jira - JiraSendFindingsResponseError: Failed to send the findings to Jira - JiraTestConnectionError: Failed to test the connection + - JiraBasicAuthError: Failed to authenticate using basic auth + - JiraInvalidParameterError: The provided parameters in Init are invalid Usage: jira = Jira( @@ -98,6 +102,10 @@ class Jira: _client_id: str = None _client_secret: str = None _access_token: str = None + _user_mail: str = None + _api_token: str = None + _domain: str = None + _using_basic_auth: bool = False _refresh_token: str = None _expiration_date: int = None _cloud_id: str = None @@ -120,14 +128,31 @@ class Jira: redirect_uri: str = None, client_id: str = None, client_secret: str = None, + user_mail: str = None, + api_token: str = None, + domain: str = None, ): self._redirect_uri = redirect_uri self._client_id = client_id self._client_secret = client_secret + self._user_mail = user_mail + self._api_token = api_token + self._domain = domain 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) + # If the client mail, API token and site name are present, use basic auth + if user_mail and api_token and domain: + self._using_basic_auth = True + self.get_basic_auth() + # If the redirect URI, client ID and client secret are present, use auth code flow + elif redirect_uri and client_id and client_secret: + auth_url = self.auth_code_url() + authorization_code = self.input_authorization_code(auth_url) + self.get_auth(authorization_code) + else: + init_error = "Failed to initialize Jira object, missing parameters." + raise JiraInvalidParameterError( + message=init_error, file=os.path.basename(__file__) + ) @property def redirect_uri(self): @@ -157,6 +182,10 @@ class Jira: def scopes(self): return self._scopes + @property + def using_basic_auth(self): + return self._using_basic_auth + def get_params(self, state_encoded): return { **self.PARAMS_TEMPLATE, @@ -209,6 +238,29 @@ class Jira: """ return (datetime.now() + timedelta(seconds=seconds)).isoformat() + def get_basic_auth(self) -> None: + """Get the access token using the mail and API token. + + Returns: + - None + + Raises: + - JiraBasicAuthError: Failed to authenticate using basic auth + """ + try: + user_string = f"{self._user_mail}:{self._api_token}" + self._access_token = base64.b64encode(user_string.encode("utf-8")).decode( + "utf-8" + ) + self._cloud_id = self.get_cloud_id(self._access_token, domain=self._domain) + except Exception as e: + message_error = f"Failed to get auth using basic auth: {e}" + logger.error(message_error) + raise JiraBasicAuthError( + message=message_error, + file=os.path.basename(__file__), + ) + def get_auth(self, auth_code: str = None) -> None: """Get the access token and refresh token @@ -269,11 +321,12 @@ class Jira: file=os.path.basename(__file__), ) - def get_cloud_id(self, access_token: str = None) -> str: + def get_cloud_id(self, access_token: str = None, domain: str = None) -> str: """Get the cloud ID from Jira Args: - access_token: The access token from Jira + - domain: The site name from Jira Returns: - str: The cloud ID @@ -284,8 +337,17 @@ class Jira: - 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 self._using_basic_auth: + headers = {"Authorization": f"Basic {access_token}"} + response = requests.get( + f"https://{domain}.atlassian.net/_edge/tenant_info", + headers=headers, + ) + response = response.json() + return response.get("cloudId") + else: + headers = {"Authorization": f"Bearer {access_token}"} + response = requests.get(self.API_TOKEN_URL, headers=headers) if response.status_code == 200: resources = response.json() @@ -326,6 +388,10 @@ class Jira: - JiraGetAccessTokenError: Failed to get the access token """ try: + # If using basic auth, return the access token + if self._using_basic_auth: + return self._access_token + if self.auth_expiration and datetime.now() < datetime.fromisoformat( self.auth_expiration ): @@ -392,6 +458,9 @@ class Jira: redirect_uri: str = None, client_id: str = None, client_secret: str = None, + user_mail: str = None, + api_token: str = None, + domain: str = None, raise_on_exception: bool = True, ) -> Connection: """Test the connection to Jira @@ -400,6 +469,9 @@ class Jira: - redirect_uri: The redirect URI - client_id: The client ID - client_secret: The client secret + - user_mail: The client mail + - api_token: The API token + - domain: The site name - raise_on_exception: Whether to raise an exception or not Returns: @@ -417,13 +489,20 @@ class Jira: redirect_uri=redirect_uri, client_id=client_id, client_secret=client_secret, + user_mail=user_mail, + api_token=api_token, + domain=domain, ) access_token = jira.get_access_token() if not access_token: return ValueError("Failed to get access token") - headers = {"Authorization": f"Bearer {access_token}"} + if jira.using_basic_auth: + headers = {"Authorization": f"Basic {access_token}"} + else: + 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, @@ -461,6 +540,13 @@ class Jira: if raise_on_exception: raise auth_error return Connection(error=auth_error) + except JiraBasicAuthError as basic_auth_error: + logger.error( + f"{basic_auth_error.__class__.__name__}[{basic_auth_error.__traceback__.tb_lineno}]: {basic_auth_error}" + ) + if raise_on_exception: + raise basic_auth_error + return Connection(error=basic_auth_error) except Exception as error: logger.error(f"Failed to test connection: {error}") if raise_on_exception: @@ -489,7 +575,11 @@ class Jira: if not access_token: return ValueError("Failed to get access token") - headers = {"Authorization": f"Bearer {access_token}"} + if self._using_basic_auth: + headers = {"Authorization": f"Basic {access_token}"} + else: + 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, @@ -500,7 +590,7 @@ class Jira: projects = { project["key"]: project["name"] for project in response.json() } - if len(projects) == 0: + if projects == {}: # If no projects are found logger.error("No projects found") raise JiraNoProjectsError( message="No projects found in Jira", @@ -555,7 +645,11 @@ class Jira: file=os.path.basename(__file__), ) - headers = {"Authorization": f"Bearer {access_token}"} + if self._using_basic_auth: + headers = {"Authorization": f"Basic {access_token}"} + else: + 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, @@ -1109,10 +1203,17 @@ class Jira: raise JiraInvalidIssueTypeError( message="The issue type is invalid", file=os.path.basename(__file__) ) - headers = { - "Authorization": f"Bearer {access_token}", - "Content-Type": "application/json", - } + + if self._using_basic_auth: + headers = { + "Authorization": f"Basic {access_token}", + "Content-Type": "application/json", + } + else: + headers = { + "Authorization": f"Bearer {access_token}", + "Content-Type": "application/json", + } for finding in findings: status_color = self.get_color_from_status(finding.status.value) diff --git a/tests/lib/outputs/jira/jira_test.py b/tests/lib/outputs/jira/jira_test.py index 238f433894..e4cc7fbc0b 100644 --- a/tests/lib/outputs/jira/jira_test.py +++ b/tests/lib/outputs/jira/jira_test.py @@ -1,3 +1,4 @@ +import base64 from datetime import datetime, timedelta from unittest.mock import MagicMock, PropertyMock, patch from urllib.parse import parse_qs, urlparse @@ -7,6 +8,7 @@ from freezegun import freeze_time from prowler.lib.outputs.jira.exceptions.exceptions import ( JiraAuthenticationError, + JiraBasicAuthError, JiraCreateIssueError, JiraGetAvailableIssueTypesError, JiraGetCloudIDError, @@ -39,6 +41,22 @@ class TestJiraIntegration: client_secret=self.client_secret, ) + @pytest.fixture(autouse=True) + @patch.object(Jira, "get_basic_auth", return_value=None) + def setup_basic_auth(self, mock_get_basic_auth): + # To disable vulture + mock_get_basic_auth = mock_get_basic_auth + + self.user_mail = "test_user_mail" + self.api_token = "test_api_token" + self.domain = "test_domain" + + self.jira_integration_basic_auth = Jira( + user_mail=self.user_mail, + api_token=self.api_token, + domain=self.domain, + ) + @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""" @@ -68,6 +86,31 @@ class TestJiraIntegration: assert query_params["response_type"][0] == "code" assert query_params["prompt"][0] == "consent" + @patch.object(Jira, "get_cloud_id", return_value="test_cloud_id") + def test_get_auth_successful_basic_auth(self, mock_get_cloud_id): + """Test successful token retrieval in get_basic_auth.""" + # To disable vulture + mock_get_cloud_id = mock_get_cloud_id + + self.jira_integration_basic_auth.get_basic_auth() + + user_string = "test_user_mail:test_api_token" + user_string_base64 = base64.b64encode(user_string.encode("utf-8")).decode( + "utf-8" + ) + + assert self.jira_integration_basic_auth._access_token == user_string_base64 + assert self.jira_integration_basic_auth._cloud_id == "test_cloud_id" + + @patch.object(Jira, "get_cloud_id", side_effect=Exception("Connection error")) + def test_get_auth_error_basic_auth(self, mock_get_cloud_id): + """Test successful token retrieval in get_basic_auth.""" + # To disable vulture + mock_get_cloud_id = mock_get_cloud_id + + with pytest.raises(JiraBasicAuthError): + self.jira_integration_basic_auth.get_basic_auth() + @freeze_time(TEST_DATETIME) @patch("prowler.lib.outputs.jira.jira.requests.post") @patch.object(Jira, "get_cloud_id", return_value="test_cloud_id") @@ -276,6 +319,33 @@ class TestJiraIntegration: assert connection.is_connected assert connection.error is None + @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_basic_auth", return_value=None) + @patch("prowler.lib.outputs.jira.jira.requests.get") + def test_test_connection_successful_basic_auth( + 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_basic_auth.test_connection( + user_mail=self.user_mail, + api_token=self.api_token, + domain=self.domain, + ) + + assert connection.is_connected + assert connection.error is None + @patch.object( Jira, "get_access_token", @@ -293,6 +363,23 @@ class TestJiraIntegration: client_secret=self.client_secret, ) + @patch.object( + Jira, + "get_cloud_id", + side_effect=JiraBasicAuthError("Failed to authenticate with Jira"), + ) + def test_test_connection_failed_basic_auth(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(JiraBasicAuthError): + self.jira_integration_basic_auth.test_connection( + user_mail=self.user_mail, + api_token=self.api_token, + domain=self.domain, + ) + @patch.object(Jira, "get_access_token", return_value="valid_access_token") @patch.object( Jira, "cloud_id", new_callable=PropertyMock, return_value="test_cloud_id"