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

Author: pedrooot

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 @@ def __init__(self, file=None, original_exception=None, message=None):
         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.outputs.finding import Finding
 from prowler.lib.outputs.jira.exceptions.exceptions import (
     JiraAuthenticationError,
+    JiraBasicAuthError,
     JiraCreateIssueError,
     JiraGetAccessTokenError,
     JiraGetAuthResponseError,
@@ -21,6 +22,7 @@
     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 @@ def __init__(
         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 @@ def cloud_id(self, value):
     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 @@ def get_timestamp_from_seconds(seconds: int) -> datetime:
         """
         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 @@ def get_auth(self, auth_code: str = None) -> None:
                 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 @@ def get_cloud_id(self, access_token: str = None) -> str:
             - 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 @@ def get_access_token(self) -> str:
             - 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 @@ def test_connection(
         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 @@ def test_connection(
             - 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 @@ def test_connection(
                 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 @@ def test_connection(
             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 @@ def get_projects(self) -> Dict[str, str]:
             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 @@ def get_projects(self) -> Dict[str, str]:
                 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 @@ def get_available_issue_types(self, project_key: str = None) -> list[str]:
                     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 @@ def send_findings(
                 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)