Fix/propegate url analysis additional parameters (#50)

* fix: propagate additional parameters to URL analysis

* fea: add from_analysis_id to sub_analysis

* add analysis summary meadata function

* fix: remove gene count software types on malicious summary report

Co-authored-by: Matan Yechiel <matany90>

CHANGES

@@ -1,3 +1,10 @@
+1.9.0
+-------
+- Rename exception to have Error suffix
+- Add `SubAnalysis.from_analysis_id` to properly initialize SubAnalysis without the composed analysis
+- Fix URL analysis additional parameters propagation
+- Add File analysis summary metadata function
+
 1.8.3
 -------
 - add extraction info to sub analysis

intezer_sdk/__init__.py

@@ -1 +1 @@
-__version__ = '1.8.3'
+__version__ = '1.9.0'

intezer_sdk/analysis.py

@@ -226,7 +226,7 @@ def _query_status_from_api(self) -> Response:
         return self._api.get_url_analysis_response(self.analysis_id, False)
 
     def _send_analyze_to_api(self, **additional_parameters) -> str:
-        return self._api.analyze_url(self.url)
+        return self._api.analyze_url(self.url, **additional_parameters)
 
     @property
     def downloaded_file_analysis(self) -> Optional[FileAnalysis]:
@@ -251,6 +251,6 @@ def get_url_analysis_by_id(analysis_id: str, api: IntezerApi = None) -> Optional
 def _assert_analysis_status(response: dict):
     if response['status'] in (consts.AnalysisStatusCode.IN_PROGRESS.value,
                               consts.AnalysisStatusCode.QUEUED.value):
-        raise errors.AnalysisIsStillRunning()
+        raise errors.AnalysisIsStillRunningError()
     if response['status'] == consts.AnalysisStatusCode.FAILED.value:
         raise errors.AnalysisFailedError()

intezer_sdk/api.py

@@ -391,7 +391,7 @@ def _set_access_token(self, api_key: str):
                                  verify=self._verify_ssl)
 
         if response.status_code in (HTTPStatus.UNAUTHORIZED, HTTPStatus.BAD_REQUEST):
-            raise errors.InvalidApiKey(response)
+            raise errors.InvalidApiKeyError(response)
         if response.status_code != HTTPStatus.OK:
             raise_for_status(response)
 
@@ -445,9 +445,9 @@ def _assert_analysis_response_status_code(response: Response):
         if response.status_code == HTTPStatus.NOT_FOUND:
             raise errors.HashDoesNotExistError(response)
         elif response.status_code == HTTPStatus.CONFLICT:
-            raise errors.AnalysisIsAlreadyRunning(response)
+            raise errors.AnalysisIsAlreadyRunningError(response)
         elif response.status_code == HTTPStatus.FORBIDDEN:
-            raise errors.InsufficientQuota(response)
+            raise errors.InsufficientQuotaError(response)
         elif response.status_code == HTTPStatus.BAD_REQUEST:
             data = response.json()
             error = data.get('error', '')
@@ -472,14 +472,14 @@ def _get_index_id_from_response(response: Response):
 
     def assert_on_premise_above_v21_11(self):
         if self.on_premise_version and self.on_premise_version <= OnPremiseVersion.V21_11:
-            raise errors.UnsupportedOnPremiseVersion('This endpoint is not available yet on this on premise')
+            raise errors.UnsupportedOnPremiseVersionError('This endpoint is not available yet on this on premise')
 
 
 def get_global_api() -> IntezerApi:
     global _global_api
 
     if not _global_api:
-        raise errors.GlobalApiIsNotInitialized()
+        raise errors.GlobalApiIsNotInitializedError()
 
     return _global_api
 

intezer_sdk/base_analysis.py

@@ -68,7 +68,7 @@ def send(self,
              wait_timeout: Optional[datetime.timedelta] = None,
              **additional_parameters) -> None:
         if self.analysis_id:
-            raise errors.AnalysisHasAlreadyBeenSent()
+            raise errors.AnalysisHasAlreadyBeenSentError()
 
         self.analysis_id = self._send_analyze_to_api(**additional_parameters)
 
@@ -101,7 +101,7 @@ def check_status(self) -> consts.AnalysisStatusCode:
 
     def result(self) -> dict:
         if self._is_analysis_running():
-            raise errors.AnalysisIsStillRunning()
+            raise errors.AnalysisIsStillRunningError()
         if not self._report:
             raise errors.ReportDoesNotExistError()
 
@@ -117,6 +117,6 @@ def set_report(self, report: dict):
 
     def _assert_analysis_finished(self):
         if self._is_analysis_running():
-            raise errors.AnalysisIsStillRunning()
+            raise errors.AnalysisIsStillRunningError()
         if self.status != consts.AnalysisStatusCode.FINISH:
             raise errors.IntezerError('Analysis not finished successfully')

intezer_sdk/errors.py

@@ -13,10 +13,13 @@ class IntezerError(Exception):
     pass
 
 
-class UnsupportedOnPremiseVersion(IntezerError):
+class UnsupportedOnPremiseVersionError(IntezerError):
     pass
 
 
+UnsupportedOnPremiseVersion = UnsupportedOnPremiseVersionError
+
+
 class ServerError(IntezerError):
     def __init__(self, message: str, response: requests.Response):
         self.response = response
@@ -26,16 +29,22 @@ def __init__(self, message: str, response: requests.Response):
         super().__init__(message)
 
 
-class AnalysisHasAlreadyBeenSent(IntezerError):
+class AnalysisHasAlreadyBeenSentError(IntezerError):
     def __init__(self):
-        super(AnalysisHasAlreadyBeenSent, self).__init__('Analysis already been sent')
+        super().__init__('Analysis already been sent')
+
 
+AnalysisHasAlreadyBeenSent = AnalysisHasAlreadyBeenSentError
 
-class IndexHasAlreadyBeenSent(IntezerError):
+
+class IndexHasAlreadyBeenSentError(IntezerError):
     def __init__(self):
         super().__init__('Index already been sent')
 
 
+IndexHasAlreadyBeenSent = IndexHasAlreadyBeenSentError
+
+
 class FamilyNotFoundError(IntezerError):
     def __init__(self, family_id: str):
         super().__init__('Family not found: {}'.format(family_id))
@@ -51,41 +60,67 @@ def __init__(self):
         super().__init__('Report was not found')
 
 
-class AnalysisIsAlreadyRunning(ServerError):
+class AnalysisIsAlreadyRunningError(ServerError):
     def __init__(self, response: requests.Response):
         super().__init__('Analysis already running', response)
 
 
-class InsufficientQuota(ServerError):
+AnalysisIsAlreadyRunning = AnalysisIsAlreadyRunningError
+
+
+class InsufficientQuotaError(ServerError):
     def __init__(self, response: requests.Response):
         super().__init__('Insufficient quota', response)
 
 
-class GlobalApiIsNotInitialized(IntezerError):
+InsufficientQuota = InsufficientQuotaError
+
+
+class GlobalApiIsNotInitializedError(IntezerError):
     def __init__(self):
         super().__init__('Global API is not initialized')
 
 
-class AnalysisIsStillRunning(IntezerError):
+GlobalApiIsNotInitialized = GlobalApiIsNotInitializedError
+
+
+class AnalysisIsStillRunningError(IntezerError):
     def __init__(self):
         super().__init__('Analysis is still running')
 
 
+AnalysisIsStillRunning = AnalysisIsStillRunningError
+
+
 class AnalysisFailedError(IntezerError):
     def __init__(self):
         super().__init__('Analysis failed')
 
 
-class InvalidApiKey(ServerError):
+class InvalidApiKeyError(ServerError):
     def __init__(self, response: requests.Response):
         super().__init__('Invalid api key', response)
 
 
-class IndexFailed(ServerError):
+InvalidApiKey = InvalidApiKeyError
+
+
+class IndexFailedError(ServerError):
     def __init__(self, response: requests.Response):
         super().__init__('Index operation failed', response)
 
 
-class SubAnalysisOperationStillRunning(IntezerError):
+IndexFailed = IndexFailedError
+
+
+class SubAnalysisOperationStillRunningError(IntezerError):
     def __init__(self, operation):
-        super(SubAnalysisOperationStillRunning, self).__init__('{} is still running'.format(operation))
+        super().__init__('{} is still running'.format(operation))
+
+
+SubAnalysisOperationStillRunning = SubAnalysisOperationStillRunningError
+
+
+class SubAnalysisNotFoundError(IntezerError):
+    def __init__(self, analysis_id: str):
+        super().__init__('analysis {} is not found'.format(analysis_id))

intezer_sdk/index.py

@@ -31,7 +31,7 @@ def __init__(self,
 
     def send(self, wait: typing.Union[bool, int] = False):
         if self.index_id:
-            raise errors.IndexHasAlreadyBeenSent()
+            raise errors.IndexHasAlreadyBeenSentError()
 
         if self._sha256:
             self.index_id = self._api.index_by_sha256(self._sha256, self._index_as, self._family_name)
@@ -70,7 +70,7 @@ def check_status(self):
         response = self._api.get_index_response(self.index_id)
         if response.status_code == HTTPStatus.OK:
             if response.json()['status'] == 'failed':
-                raise errors.IndexFailed(response)
+                raise errors.IndexFailedError(response)
             else:
                 self.status = consts.IndexStatusCode.FINISH
         elif response.status_code == HTTPStatus.ACCEPTED:

intezer_sdk/operation.py

@@ -28,7 +28,7 @@ def get_result(self):
                 self.result = operation_result.json()['result']
                 self.status = AnalysisStatusCode.FINISH
             else:
-                raise errors.SubAnalysisOperationStillRunning('operation')
+                raise errors.SubAnalysisOperationStillRunningError('operation')
         return self.result
 
     def wait_for_completion(self,

intezer_sdk/sub_analysis.py

@@ -2,6 +2,7 @@
 from typing import Optional
 from typing import Union
 
+from intezer_sdk import errors
 from intezer_sdk.api import IntezerApi
 from intezer_sdk.api import get_global_api
 from intezer_sdk.consts import AnalysisStatusCode
@@ -18,14 +19,50 @@ def __init__(self,
                  api: IntezerApi = None):
         self.composed_analysis_id = composed_analysis_id
         self.analysis_id = analysis_id
-        self.sha256 = sha256
-        self.source = source
-        self.extraction_info = extraction_info
+        self._sha256 = sha256
+        self._source = source
+        self._extraction_info = extraction_info
         self._api = api or get_global_api()
         self._code_reuse = None
         self._metadata = None
         self._operations = {}
 
+    @classmethod
+    def from_analysis_id(cls,
+                         analysis_id: str,
+                         composed_analysis_id: str,
+                         lazy_load=True,
+                         api: IntezerApi = None) -> Optional['SubAnalysis']:
+        sub_analysis = cls(analysis_id, composed_analysis_id, '', '', None, api)
+        if not lazy_load:
+            try:
+                sub_analysis._init_sub_analysis_from_parent()
+            except errors.SubAnalysisNotFoundError:
+                return None
+        return sub_analysis
+
+    @property
+    def sha256(self) -> str:
+        if not self._sha256:
+            self._init_sub_analysis_from_parent()
+
+        return self._sha256
+
+    @property
+    def source(self) -> str:
+        if not self._source:
+            self._init_sub_analysis_from_parent()
+
+        return self._source
+
+    @property
+    def extraction_info(self) -> Optional[dict]:
+        # Since extraction_info could be none, we check if the sha256 was provided, signaling we already fetch it
+        if not self._sha256:
+            self._init_sub_analysis_from_parent()
+
+        return self._extraction_info
+
     @property
     def code_reuse(self):
         if self._code_reuse is None:
@@ -38,6 +75,18 @@ def metadata(self):
             self._metadata = self._api.get_sub_analysis_metadata_by_id(self.composed_analysis_id, self.analysis_id)
         return self._metadata
 
+    def _init_sub_analysis_from_parent(self):
+        sub_analyses = self._api.get_sub_analyses_by_id(self.composed_analysis_id)
+        sub_analysis = next((
+            sub_analysis for sub_analysis in sub_analyses if sub_analysis['sub_analysis_id'] == self.analysis_id),
+            None)
+        if not sub_analysis:
+            raise errors.SubAnalysisNotFoundError(self.analysis_id)
+
+        self._sha256 = sub_analysis['sha256']
+        self._source = sub_analysis['source']
+        self._extraction_info = sub_analysis.get('extraction_info')
+
     def find_related_files(self,
                            family_id: str,
                            wait: Union[bool, int] = False,

intezer_sdk/util.py

@@ -1,5 +1,6 @@
 import collections
 import itertools
+from typing import Dict
 from typing import List
 from typing import Optional
 from typing import Tuple
@@ -25,6 +26,51 @@ def _get_title(short: bool) -> str:
             '=========================\n\n')
 
 
+def get_analysis_summary_metadata(analysis: FileAnalysis, use_hash_link=False) -> Dict[str, any]:
+    result = analysis.result()
+    verdict = result['verdict'].lower()
+    sub_verdict = result['sub_verdict'].lower()
+    analysis_url = f"{ANALYZE_URL}/files/{result['sha256']}?private=true" if use_hash_link else result['analysis_url']
+    main_family = None
+    gene_count = None
+    iocs = None
+    dynamic_ttps = None
+    related_samples_unique_count = None
+
+    software_type_priorities_by_verdict = {
+        'malicious': [],
+        'trusted': ['application', 'library', 'interpreter', 'installer'],
+        'suspicious': ['administration_tool', 'packer']
+    }
+
+    software_type_priorities = software_type_priorities_by_verdict.get(verdict)
+    if software_type_priorities:
+        main_family, gene_count = get_analysis_family(analysis, software_type_priorities)
+
+    if verdict in ('malicious', 'suspicious'):
+        iocs = analysis.iocs
+        dynamic_ttps = analysis.dynamic_ttps
+
+    related_samples = [sub_analysis.get_account_related_samples(wait=True) for sub_analysis in
+                       analysis.get_sub_analyses()]
+    if related_samples:
+        related_samples_unique_count = len({analysis['analysis']['sha256'] for analysis in
+                                            itertools.chain.from_iterable(
+                                                sample.result['related_samples'] for sample in related_samples
+                                                if sample is not None)})
+
+    return {
+        'verdict': verdict,
+        'sub_verdict': sub_verdict,
+        'analysis_url': analysis_url,
+        'main_family': main_family,
+        'gene_count': gene_count,
+        'iocs': iocs,
+        'dynamic_ttps': dynamic_ttps,
+        'related_samples_unique_count': related_samples_unique_count
+    }
+
+
 def get_analysis_summary(analysis: FileAnalysis,
                          no_emojis: bool = False,
                          short: bool = False,
@@ -42,7 +88,7 @@ def get_analysis_summary(analysis: FileAnalysis,
         emoji = get_emoji(verdict)
 
     if verdict == 'malicious':
-        main_family, gene_count = get_analysis_family(analysis, ['malware', 'malicious_packer'])
+        main_family, gene_count = get_analysis_family(analysis, [])
     elif verdict == 'trusted':
         main_family, gene_count = get_analysis_family(analysis, ['application', 'library', 'interpreter', 'installer'])
     elif verdict == 'suspicious':