Feature/offline endpoint scan (#75)

* feat(offline-endpoint-scan): Add EndpointScanApi and expand EndpointAnalysis to support uploading offline scan files

CHANGES

@@ -1,3 +1,8 @@
+1.15.0
+-------
+- Support for offline endpoint scan uploading in 'EndpointAnalysis'.
+
+
 1.14.4
 -------
 - Add analysis time to analysis object

examples/upload_offline_endpoint_scan.py

@@ -0,0 +1,15 @@
+import sys
+from pprint import pprint
+
+from intezer_sdk import api
+from intezer_sdk.endpoint_analysis import EndpointAnalysis
+
+
+def send_file_with_wait(offline_scan_directory: str):
+    api.set_global_api('api-key')
+    analysis = EndpointAnalysis(offline_scan_directory=offline_scan_directory)
+    analysis.send(wait=True)
+    pprint(analysis.result())
+
+if __name__ == '__main__':
+    send_file_with_wait(*sys.argv[1:])

intezer_sdk/__init__.py

@@ -1 +1 @@
-__version__ = '1.14.4'
+__version__ = '1.15.0'

intezer_sdk/_endpoint_analysis_api.py

@@ -0,0 +1,98 @@
+import gzip
+from typing import List
+from urllib.parse import urlparse
+
+from intezer_sdk.api import raise_for_status
+
+from intezer_sdk.api import IntezerProxy
+
+
+class EndpointScanApi:
+    def __init__(self,
+                 scan_id: str,
+                 base_api: IntezerProxy):
+        self.base_api = base_api
+        if not scan_id:
+            raise ValueError('scan_id must be provided')
+        self.scan_id = scan_id
+        api_base = f'https://{urlparse(base_api.base_url).netloc}'
+        self.base_url = f'{api_base}/scans/scans/{scan_id}'
+
+    def request_with_refresh_expired_access_token(self, *args, **kwargs):
+        return self.base_api.request_with_refresh_expired_access_token(base_url=self.base_url, *args, **kwargs)
+
+    def send_host_info(self, host_info: dict):
+        response = self.request_with_refresh_expired_access_token(path='/host-info',
+                                                                  data=host_info,
+                                                                  method='POST')
+        raise_for_status(response)
+
+    def send_processes_info(self, processes_info: dict):
+        response = self.request_with_refresh_expired_access_token(path='/processes-info',
+                                                                  data=processes_info,
+                                                                  method='POST')
+        raise_for_status(response)
+
+    def send_loaded_modules_info(self, pid, loaded_modules_info: dict):
+        response = self.request_with_refresh_expired_access_token(path=f'/processes/{pid}/loaded-modules-info',
+                                                                  data=loaded_modules_info,
+                                                                  method='POST')
+        raise_for_status(response)
+
+    def send_injected_modules_info(self, injected_module_list: dict):
+        response = self.request_with_refresh_expired_access_token(path='/injected-modules-info',
+                                                                  data=injected_module_list,
+                                                                  method='POST')
+        raise_for_status(response)
+
+    def send_scheduled_tasks_info(self, scheduled_tasks_info: dict):
+        response = self.request_with_refresh_expired_access_token(path='/scheduled-tasks-info',
+                                                                  data=scheduled_tasks_info,
+                                                                  method='POST')
+        raise_for_status(response)
+
+    def send_file_module_differences(self, file_module_differences: dict):
+        response = self.request_with_refresh_expired_access_token(path='/file-module-differences',
+                                                                  data=file_module_differences,
+                                                                  method='POST')
+        raise_for_status(response)
+
+    def send_files_info(self, files_info: dict) -> List[str]:
+        """
+        :param files_info: endpoint scan files info
+        :return: list of file hashes to upload
+        """
+        response = self.request_with_refresh_expired_access_token(path='/files-info',
+                                                                  data=files_info,
+                                                                  method='POST')
+        raise_for_status(response)
+        return response.json()['result']
+
+    def send_memory_module_dump_info(self, memory_modules_info: dict) -> List[str]:
+        """
+        :param memory_modules_info: endpoint scan memory modules info
+        :return: list of file hashes to upload
+        """
+        response = self.request_with_refresh_expired_access_token(path='/memory-module-dumps-info',
+                                                                  data=memory_modules_info,
+                                                                  method='POST')
+        raise_for_status(response)
+        return response.json()['result']
+
+    def upload_collected_binary(self, file_path: str, collected_from: str):
+        with open(file_path, 'rb') as file_to_upload:
+            file_data = file_to_upload.read()
+            compressed_data = gzip.compress(file_data, compresslevel=9)
+            response = self.request_with_refresh_expired_access_token(
+                path=f'/{collected_from}/collected-binaries',
+                data=compressed_data,
+                headers={'Content-Type': 'application/octet-stream', 'Content-Encoding': 'gzip'},
+                method='POST')
+
+        raise_for_status(response)
+
+    def end_scan(self, scan_summary: dict):
+        response = self.request_with_refresh_expired_access_token(path='/end',
+                                                                  data=scan_summary,
+                                                                  method='POST')
+        raise_for_status(response)

intezer_sdk/analysis.py

@@ -16,17 +16,16 @@
 from intezer_sdk._util import deprecated
 from intezer_sdk.api import IntezerApi
 from intezer_sdk.api import get_global_api
-from intezer_sdk.base_analysis import BaseAnalysis
+from intezer_sdk.base_analysis import Analysis
 from intezer_sdk.sub_analysis import SubAnalysis
 
 logger = logging.getLogger(__name__)
 
 
-class FileAnalysis(BaseAnalysis):
+class FileAnalysis(Analysis):
     """
     FileAnalysis is a class for analyzing files. It is a subclass of the BaseAnalysis class and requires an API connection to Intezer.
     """
-
     def __init__(self,
                  file_path: str = None,
                  file_hash: str = None,
@@ -257,10 +256,7 @@ def get_analysis_by_id(analysis_id: str, api: IntezerApi = None) -> Optional[Fil
     return get_file_analysis_by_id(analysis_id, api)
 
 
-Analysis = FileAnalysis
-
-
-class UrlAnalysis(BaseAnalysis):
+class UrlAnalysis(Analysis):
     def __init__(self, url: Optional[str] = None, api: IntezerApi = None):
         super().__init__(api)
         self._api.assert_any_on_premise()

intezer_sdk/api.py

@@ -35,44 +35,46 @@ def raise_for_status(response: requests.Response,
     if statuses_to_ignore and response.status_code in statuses_to_ignore:
         return
     elif allowed_statuses and response.status_code not in allowed_statuses:
-        http_error_msg = '%s Custom Error: %s for url: %s' % (response.status_code, reason, response.url)
+        http_error_msg = f'{response.status_code} Custom Error: {reason} for url: {response.url}'
     elif 400 <= response.status_code < 500:
         if response.status_code != 400:
-            http_error_msg = '%s Client Error: %s for url: %s' % (response.status_code, reason, response.url)
+            http_error_msg = f'{response.status_code} Client Error: {reason} for url: {response.url}'
         else:
             # noinspection PyBroadException
             try:
                 error = response.json()
-                http_error_msg = '\n'.join(['{}:{}.'.format(key, value) for key, value in error['message'].items()])
+                http_error_msg = '\n'.join([f'{key}:{value}.' for key, value in error['message'].items()])
             except Exception:
-                http_error_msg = '%s Client Error: %s for url: %s' % (response.status_code, reason, response.url)
+                http_error_msg = f'{response.status_code} Client Error: {reason} for url: {response.url}'
     elif 500 <= response.status_code < 600:
-        http_error_msg = '%s Server Error: %s for url: %s' % (response.status_code, reason, response.url)
+        http_error_msg = f'{response.status_code} Server Error: {reason} for url: {response.url}'
 
     if http_error_msg:
         # noinspection PyBroadException
         try:
             data = response.json()
-            http_error_msg = '%s, server returns %s, details: %s' % (http_error_msg, data['error'], data.get('details'))
+            http_error_msg = f'{http_error_msg}, server returns {data["error"]}, details: {data.get("details")}'
         except Exception:
             pass
 
         raise requests.HTTPError(http_error_msg, response=response)
 
 
-class IntezerApi:
+class IntezerProxy:
     def __init__(self,
+                 *,
                  api_version: str = None,
                  api_key: str = None,
                  base_url: str = None,
                  verify_ssl: bool = True,
                  on_premise_version: OnPremiseVersion = None,
                  user_agent: str = None):
         self.full_url = base_url + api_version
+        self.base_url = base_url
         self.api_key = api_key
         self._access_token = None
         self._session = None
-        self._verify_ssl = verify_ssl
+        self.verify_ssl = verify_ssl
         self.on_premise_version = on_premise_version
         if user_agent:
             user_agent = f'{consts.USER_AGENT}/{user_agent}'
@@ -86,23 +88,35 @@ def _request(self,
                  data: dict = None,
                  headers: dict = None,
                  files: dict = None,
-                 stream: bool = None) -> Response:
+                 stream: bool = None,
+                 base_url: str = None) -> Response:
         if not self._session:
             self.set_session()
 
+        url = f'{base_url}{path}' if base_url else f'{self.full_url}{path}'
+
         if files:
             response = self._session.request(
                 method,
-                self.full_url + path,
+                url,
                 files=files,
                 data=data or {},
                 headers=headers or {},
                 stream=stream
             )
+        elif isinstance(data, bytes):
+            response = self._session.request(
+                method,
+                url,
+                files=files,
+                data=data,
+                headers=headers or {},
+                stream=stream
+            )
         else:
             response = self._session.request(
                 method,
-                self.full_url + path,
+                url,
                 json=data or {},
                 headers=headers,
                 stream=stream
@@ -116,16 +130,54 @@ def request_with_refresh_expired_access_token(self,
                                                   data: dict = None,
                                                   headers: dict = None,
                                                   files: dict = None,
-                                                  stream: bool = None) -> Response:
-        response = self._request(method, path, data, headers, files)
+                                                  stream: bool = None,
+                                                  base_url: str = None) -> Response:
+        response = self._request(method, path, data, headers, files, stream, base_url=base_url)
 
         if response.status_code == HTTPStatus.UNAUTHORIZED:
             self._access_token = None
             self.set_session()
-            response = self._request(method, path, data, headers, files, stream)
+            response = self._request(method, path, data, headers, files, stream, base_url)
 
         return response
 
+    def _set_access_token(self, api_key: str):
+        response = requests.post(f'{self.full_url}/get-access-token',
+                                 json={'api_key': api_key},
+                                 verify=self.verify_ssl)
+
+        if response.status_code in (HTTPStatus.UNAUTHORIZED, HTTPStatus.BAD_REQUEST):
+            raise errors.InvalidApiKeyError(response)
+        if response.status_code != HTTPStatus.OK:
+            raise_for_status(response)
+
+        self._access_token = response.json()['result']
+
+    def set_session(self):
+        self._session = requests.session()
+        self._session.mount('https://', requests.adapters.HTTPAdapter(max_retries=3))
+        self._session.mount('http://', requests.adapters.HTTPAdapter(max_retries=3))
+        self._session.verify = self.verify_ssl
+        self._set_access_token(self.api_key)
+        self._session.headers['Authorization'] = f'Bearer {self._access_token}'
+        self._session.headers['User-Agent'] = self.user_agent
+
+
+class IntezerApi(IntezerProxy):
+    def __init__(self,
+                 api_version: str = None,
+                 api_key: str = None,
+                 base_url: str = None,
+                 verify_ssl: bool = True,
+                 on_premise_version: OnPremiseVersion = None,
+                 user_agent: str = None):
+        super().__init__(api_key=api_key,
+                         base_url=base_url,
+                         verify_ssl=verify_ssl,
+                         user_agent=user_agent,
+                         api_version=api_version,
+                         on_premise_version=on_premise_version)
+
     def analyze_by_hash(self,
                         file_hash: str,
                         disable_dynamic_unpacking: Optional[bool],
@@ -254,6 +306,17 @@ def get_endpoint_sub_analyses(self, analyses_id: str, verdicts: Optional[List[st
 
         return response.json()['sub_analyses']
 
+    def  create_endpoint_scan(self, scanner_info: dict) -> Dict[str, str]:
+        if not self.on_premise_version or self.on_premise_version > OnPremiseVersion.V22_10:
+            scanner_info['scan_type'] = consts.SCAN_TYPE_OFFLINE_ENDPOINT_SCAN
+        response = self.request_with_refresh_expired_access_token(path='scans',
+                                                                  data=scanner_info,
+                                                                  method='POST',
+                                                                  base_url=self.base_url)
+
+        raise_for_status(response)
+        return response.json()['result']
+
     def get_iocs(self, analyses_id: str) -> Optional[dict]:
         response = self.request_with_refresh_expired_access_token(path='/analyses/{}/iocs'.format(analyses_id),
                                                                   method='GET')
@@ -466,26 +529,6 @@ def get_index_response(self, index_id: str) -> Response:
 
         return response
 
-    def _set_access_token(self):
-        response = requests.post(self.full_url + '/get-access-token',
-                                 json={'api_key': self.api_key},
-                                 verify=self._verify_ssl)
-
-        if response.status_code in (HTTPStatus.UNAUTHORIZED, HTTPStatus.BAD_REQUEST):
-            raise errors.InvalidApiKeyError(response)
-        if response.status_code != HTTPStatus.OK:
-            raise_for_status(response)
-
-        self._access_token = response.json()['result']
-
-    def set_session(self):
-        self._session = requests.session()
-        self._session.mount('https://', requests.adapters.HTTPAdapter(max_retries=3))
-        self._session.verify = self._verify_ssl
-        self._set_access_token()
-        self._session.headers['Authorization'] = 'Bearer {}'.format(self._access_token)
-        self._session.headers['User-Agent'] = self.user_agent
-
     def analyze_url(self, url: str, **additional_parameters) -> Optional[str]:
         self.assert_any_on_premise()
         response = self.request_with_refresh_expired_access_token(method='POST',
@@ -598,3 +641,4 @@ def set_global_api(api_key: str = None,
                              base_url or consts.BASE_URL,
                              verify_ssl,
                              on_premise_version)
+

intezer_sdk/base_analysis.py

@@ -140,8 +140,6 @@ def _create_analysis_from_response(cls, response: Response, api: IntezerApi, ana
 
         return analysis
 
-
-class BaseAnalysis(Analysis):
     @abc.abstractmethod
     def _send_analyze_to_api(self, **additional_parameters) -> str:
         raise NotImplementedError()
@@ -162,3 +160,4 @@ def send(self,
                 self.wait_for_completion(sleep_before_first_check=True, timeout=wait_timeout)
             else:
                 self.wait_for_completion(wait, sleep_before_first_check=True, timeout=wait_timeout)
+

intezer_sdk/consts.py

@@ -17,6 +17,12 @@ class AnalysisStatusCode(enum.Enum):
     FINISHED = 'finished'
 
 
+class EndpointAnalysisEndReason(enum.Enum):
+    DONE = 'done'
+    INTERRUPTED = 'interrupted'
+    FAILED = 'failed'
+
+
 class SoftwareType(AutoName):
     ADMINISTRATION_TOOL = enum.auto()
     APPLICATION = enum.auto()
@@ -83,7 +89,8 @@ class OnPremiseVersion(enum.IntEnum):
 
 
 ANALYZE_URL = 'https://analyze.intezer.com'
-BASE_URL = '{}/api/'.format(ANALYZE_URL)
+BASE_URL = f'{ANALYZE_URL}/api/'
 API_VERSION = 'v2-0'
-USER_AGENT = 'intezer-python-sdk-{}'.format(__version__)
+USER_AGENT = f'intezer-python-sdk-{__version__}'
 CHECK_STATUS_INTERVAL = 1
+SCAN_TYPE_OFFLINE_ENDPOINT_SCAN = 'offline_endpoint_scan'