feat(jira): add basic auth method (#7233)

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
+        )

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)

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"