Update QualysFIM events and incidents API endpoints to V2 and V3

Packs/QualysFIM/Integrations/QualysFIM/QualysFIM.py

@@ -8,7 +8,6 @@
 import json
 import requests
 import dateparser
-from typing import Tuple
 
 # Disable insecure warnings
 urllib3.disable_warnings()
@@ -28,11 +27,11 @@ class Client(BaseClient):
     """
 
     def __init__(self, base_url: str, verify: bool, proxy: bool, auth: tuple):
-        headers = self.get_token_and_set_headers(base_url, auth)
+        headers = self.get_token_and_set_headers(base_url, auth, verify)
         super().__init__(base_url=base_url, verify=verify, proxy=proxy, headers=headers)
 
     @staticmethod
-    def get_token_and_set_headers(base_url: str, auth: tuple):
+    def get_token_and_set_headers(base_url: str, auth: tuple, verify: bool) -> dict:
         """
         Get JWT token by authentication and set headers.
 
@@ -41,22 +40,24 @@ def get_token_and_set_headers(base_url: str, auth: tuple):
             auth (tuple): credentials for authentication.
 
         Returns:
-             headers with token.
+            headers with token.
+
+        Raises:
+            DemistoException if authentication request was not successful.
         """
         try:
-            data = {
-                'username': auth[0],
-                'password': auth[1],
-                'token': True}
+            data = {'username': auth[0], 'password': auth[1], 'token': True}
             headers = {'ContentType': 'application/x-www-form-urlencoded'}
-            token = requests.post(url=f'{base_url}/auth',
-                                  headers=headers,
-                                  data=data).text
+            url = urljoin(base_url, '/auth')
+            auth_response = requests.post(url=url, headers=headers, data=data, verify=verify)
+            auth_response.raise_for_status()
+
+            token = auth_response.text
             return {'Authorization': f'Bearer {token}', 'content-type': 'application/json'}
-        except Exception:
-            raise ValueError('URL is not set correctly, please review URL,\n'
-                             'Read URL instructions at ? button in "Qualys API Platform URL" '
-                             'parameter')
+
+        except requests.exceptions.HTTPError:
+            raise DemistoException('Authentication failed. Verify the Qualys API Platform URL, '
+                                   'access credentials, and other connection parameters.')
 
     def incidents_list_test(self):
         """
@@ -65,8 +66,7 @@ def incidents_list_test(self):
         return:
             response (Response): API response from Qualys FIM.
         """
-        return self._http_request(method='GET', url_suffix='fim/v1/incidents/',
-                                  params={'pageSize': '1'}, resp_type='response')
+        return self.incidents_list(data={'pageSize': '1'})
 
     def events_list(self, data: dict):
         """
@@ -90,7 +90,7 @@ def get_event(self, event_id: str):
         return:
             response (Dict): API response from Qualys FIM.
         """
-        return self._http_request(method='GET', url_suffix=f'fim/v1/events/{event_id}')
+        return self._http_request(method='GET', url_suffix=f'fim/v2/events/{event_id}')
 
     def incidents_list(self, data: dict):
         """
@@ -102,8 +102,7 @@ def incidents_list(self, data: dict):
         return:
             response (Dict): API response from Qualys FIM.
         """
-        return self._http_request(method='POST', url_suffix='fim/v3/incidents/search',
-                                  json_data=data)
+        return self._http_request(method='POST', url_suffix='fim/v3/incidents/search', json_data=data)
 
     def get_incident_events(self, incident_id: str, data: dict):
         """
@@ -557,7 +556,7 @@ def list_assets_command(client: Client, args: dict):
 
 def fetch_incidents(client: Client, last_run: Dict[str, int],
                     max_fetch: str, fetch_filter: str,
-                    first_fetch_time: str) -> Tuple[Dict, List[dict]]:
+                    first_fetch_time: str) -> tuple[Dict, List[dict]]:
     """
     Fetch incidents (alerts) each minute (by default).
     Args:
@@ -635,9 +634,8 @@ def test_module(client: Client):
         'ok' if test passed, anything else will fail the test.
     """
     try:
-        result = client.incidents_list_test()
-        if result.ok:
-            return 'ok'
+        client.incidents_list_test()  # raises exception if non-okay response
+        return 'ok'
     except Exception as exception:
         error_msg = None
         if 'Authorization' in str(exception):

Packs/QualysFIM/Integrations/QualysFIM/QualysFIM.yml

@@ -399,7 +399,7 @@ script:
     - contextPath: QualysFIM.Assets.created
       description: Date the asset was created.
       type: String
-  dockerimage: demisto/python3:3.10.14.91134
+  dockerimage: demisto/python3:3.11.10.115186
   isfetch: true
   runonce: false
   script: '-'