Merge pull request #38 from MislavReversingLabs/main

MislavReversingLabs · web-flow · commit d839923927c3 · 2024-03-30T21:19:11.000+01:00
v2.5.0
diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml
@@ -22,4 +22,4 @@ jobs:
           pip install pytest requests
       - name: Test with pytest
         run: |
-          pytest
+          pytest
diff --git a/.gitignore b/.gitignore
@@ -6,3 +6,8 @@ __pycache__/
 
 .vscode/
 .DS_Store
+
+*.pyc
+.pytest_cache
+.coverage
+.idea
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -210,13 +210,20 @@ v2.4.3 (2024-02-07)
 -------------------
 
 #### Improvements
-
 - Python package dependencies are now set to the following values:
   - `requests>=2.28.2`
   - `urllib3>=1.26.14`
 
 
+v2.5.0 (2024-03-29)
+-------------------
 
-### Scheduled removals
-- **March 2024.**:
-  - `a1000.A1000.advanced_search_v2`
+#### Removals
+- **a1000** module:
+  - Removed the `a1000.A1000.advanced_search_v2` method.
+
+#### Improvements
+- Added unit tests.
+- Added CI/CD (Actions) workflows for running unit tests and publishing the package to PyPI.
+- **ticloud** module:
+  - `md5` and `sha256` can now be used in `DynamicAnalysis.get_dynamic_analysis_results` for fetching sample analysis results.
diff --git a/README.md b/README.md
@@ -1,13 +1,16 @@
 ![ReversingLabs](https://raw.githubusercontent.com/reversinglabs/reversinglabs-sdk-py3/master/logo.png)
 
-# ReversingLabsSDK
+# ReversingLabs SDK
 
-A Python SDK for ReversingLabs REST services (TitaniumCloud and appliances) - Python 3 version.
+The official Python SDK for using ReversingLabs services.
 
 The idea behind this SDK is to enable easier out-of-the-box development of software integrations and automation services that need to interact with ReversingLabs.
 
 The SDK consists of several modules, where each module represents either one ReversingLabs service, ReversingLabs appliance or the ReversingLabs TitaniumCloud.
 
+> **ReversingLabs SDK Cookbook**  
+For a simple and comprehensive guide on how to start using the ReversingLabs SDK, visit the [ReversingLabs SDK Cookbook](https://github.com/reversinglabs/reversinglabs-sdk-cookbook) 
+> and explore the provided steps and examples.
 
 ## Module: a1000
 A Python module representing the ReversingLabs A1000 malware analysis platform.
diff --git a/ReversingLabs/SDK/__init__.py b/ReversingLabs/SDK/__init__.py
@@ -0,0 +1,8 @@
+"""
+author: Mislav Sever
+
+ReversingLabs SDK
+A Python SDK for communicating with ReversingLabs services.
+"""
+
+__version__ = "2.5.0"
diff --git a/ReversingLabs/SDK/a1000.py b/ReversingLabs/SDK/a1000.py
@@ -9,7 +9,6 @@
 import requests
 import time
 from urllib import parse
-from warnings import warn
 
 from ReversingLabs.SDK.helper import ADVANCED_SEARCH_SORTING_CRITERIA, DEFAULT_USER_AGENT, RESPONSE_CODE_ERROR_MAP, \
     MD5, SHA1, SHA256, SHA512, \
@@ -1560,131 +1559,6 @@ def get_yara_cloud_retro_scan_status(self, ruleset_name):
 
         return response
 
-    def advanced_search_v2(self, query_string, ticloud=False, page_number=1, records_per_page=20, sorting_criteria=None,
-                           sorting_order="desc"):
-        """THIS METHOD IS DEPRECATED. Use advanced_search_v3 instead.
-
-        Sends a query string to the A1000 Advanced Search API v2.
-        The query string must be composed of key-value pairs separated by space.
-        A key is separated from its value by a colon symbol and no spaces.
-        For directions on how to write advanced search queries, consult the A1000 documentation.
-        If a page number is not provided, the first page of results will be returned.
-            Query string example:
-            'av-count:5 available:TRUE'
-
-            :param query_string: query string
-            :type query_string: str
-            :param ticloud: show only cloud results
-            :type ticloud: bool
-            :param page_number: page number
-            :type page_number: int
-            :param records_per_page: number of records returned per page; maximum value is 100
-            :type records_per_page: int
-            :param sorting_criteria: define the criteria used in sorting; possible values are 'sha1', 'firstseen',
-            'threatname', 'sampletype', 'filecount', 'size'
-            :type sorting_criteria: str
-            :param sorting_order: sorting order; possible values are 'desc', 'asc'
-            :type sorting_order: str
-            :return: response
-            :rtype: requests.Response
-        """
-        warn("This method is deprecated. Use advanced_search_v3 instead.", DeprecationWarning)
-
-        if not isinstance(query_string, str):
-            raise WrongInputError("The search query must be a string.")
-
-        if not isinstance(ticloud, bool):
-            raise WrongInputError("ticloud parameter must be boolean.")
-
-        if not isinstance(records_per_page, int) or not 1 <= records_per_page <= 100:
-            raise WrongInputError("records_per_page parameter must be an integer with a value "
-                                  "between 1 and 100 (included).")
-
-        url = self._url.format(endpoint=self.__ADVANCED_SEARCH_ENDPOINT_V2)
-
-        post_json = {"query": query_string, "ticloud": ticloud, "page": page_number,
-                     "records_per_page": records_per_page}
-
-        if sorting_criteria:
-            if sorting_criteria not in ADVANCED_SEARCH_SORTING_CRITERIA or sorting_order not in ("desc", "asc"):
-                raise WrongInputError("Sorting criteria must be one of the following options: {criteria}. "
-                                      "Sorting order needs to be 'desc' or 'asc'.".format(
-                                        criteria=ADVANCED_SEARCH_SORTING_CRITERIA
-                                      ))
-            sorting_expression = "{criteria} {order}".format(
-                criteria=sorting_criteria,
-                order=sorting_order
-            )
-
-            post_json["sort"] = sorting_expression
-
-        response = self.__post_request(url=url, post_json=post_json)
-
-        self.__raise_on_error(response)
-
-        return response
-
-    def advanced_search_v2_aggregated(self,  query_string, ticloud=False, max_results=5000, sorting_criteria=None,
-                                      sorting_order="desc"):
-        """THIS METHOD IS DEPRECATED. Use advanced_search_v3_aggregated instead.
-
-        Sends a query string to the A1000 Advanced Search API v2.
-        The query string must be composed of key-value pairs separated by space.
-        A key is separated from its value by a colon symbol and no spaces.
-        For directions on how to write advanced search queries, consult the A1000 documentation.
-        Paging is done automatically and results from individual
-        responses aggregated into one list and returned`.
-        The 'max_results' parameter defines the maximum desired number of results to be returned.
-            Query string example:
-            'av-count:5 available:TRUE'
-
-            :param query_string: search query - see API documentation for details on writing search queries
-            :type query_string: str
-            :param ticloud: show only cloud results
-            :type ticloud: bool
-            :param max_results: maximum results to be returned in a list; default value is 5000
-            :type max_results: int
-            :param sorting_criteria: define the criteria used in sorting; possible values are 'sha1', 'firstseen',
-            'threatname', 'sampletype', 'filecount', 'size'
-            :type sorting_criteria: str
-            :param sorting_order: sorting order; possible values are 'desc', 'asc'
-            :type sorting_order: str
-            :return: list of results
-            :rtype: list
-        """
-        warn("This method is deprecated. Use advanced_search_v3_aggregated instead.", DeprecationWarning)
-
-        if not isinstance(max_results, int):
-            raise WrongInputError("max_results parameter must be integer.")
-
-        results = []
-        next_page = 1
-        more_pages = True
-
-        while more_pages:
-            response = self.advanced_search_v2(
-                query_string=query_string,
-                ticloud=ticloud,
-                page_number=next_page,
-                records_per_page=100,
-                sorting_criteria=sorting_criteria,
-                sorting_order=sorting_order
-            )
-
-            response_json = response.json()
-
-            entries = response_json.get("rl").get("web_search_api").get("entries", [])
-            results.extend(entries)
-
-            if len(results) > max_results:
-                results = results[:max_results]
-                return results
-
-            next_page = response_json.get("rl").get("web_search_api").get("next_page", None)
-            more_pages = response_json.get("rl").get("web_search_api").get("more_pages", False)
-
-        return results
-
     def advanced_search_v3(self, query_string, ticloud=False, start_search_date=None, end_search_date=None,
                            page_number=1, records_per_page=20, sorting_criteria=None, sorting_order="desc"):
         """Sends a query string to the A1000 Advanced Search API v3.
@@ -2437,4 +2311,4 @@ def __raise_on_error(response):
         exception = RESPONSE_CODE_ERROR_MAP.get(response.status_code, None)
         if not exception:
             return
-        raise exception
+        raise exception(response_object=response)
diff --git a/ReversingLabs/SDK/helper.py b/ReversingLabs/SDK/helper.py
@@ -8,6 +8,7 @@
 import codecs
 import binascii
 from http import HTTPStatus
+from ReversingLabs.SDK import __version__
 
 
 MD5 = "md5"
@@ -22,7 +23,7 @@
     128: SHA512
 }
 
-DEFAULT_USER_AGENT = "ReversingLabs SDK v2.4.3"
+DEFAULT_USER_AGENT = f"ReversingLabs SDK v{__version__}"
 ADVANCED_SEARCH_SORTING_CRITERIA = ("sha1", "firstseen", "threatname", "sampletype", "filecount", "size")
 
 
diff --git a/ReversingLabs/SDK/ticloud.py b/ReversingLabs/SDK/ticloud.py
@@ -3127,8 +3127,8 @@ class DynamicAnalysis(TiCloudAPI):
 
     __DETONATE_ENDPOINT = "/api/dynamic/analysis/analyze/v1/query/json"
     __DETONATE_ARCHIVE_ENDPOINT = "/api/dynamic/analysis/analyze/v1/archive/query/json"
-    __GET_FILE_RESULTS = "/api/dynamic/analysis/report/v1/query/sha1"
-    __GET_ARCHIVE_RESULTS_ENDPOINT = "/api/dynamic/analysis/report/v1/archive/query/sha1"
+    __GET_FILE_RESULTS = "/api/dynamic/analysis/report/v1/query/{hash_type}"
+    __GET_ARCHIVE_RESULTS_ENDPOINT = "/api/dynamic/analysis/report/v1/archive/query/{hash_type}"
     __GET_URL_RESULTS_BASE64 = "/api/dynamic/analysis/report/v1/query/url/base64"
     __GET_URL_RESULTS_SHA1 = "/api/dynamic/analysis/report/v1/query/url/sha1"
 
@@ -3261,7 +3261,7 @@ def get_dynamic_analysis_results(self, sample_hash=None, url=None, url_sha1=None
         if sample_hash:
             validate_hashes(
                 hash_input=[sample_hash],
-                allowed_hash_types=(SHA1,)
+                allowed_hash_types=(SHA1, MD5, SHA256)
             )
             indicator = sample_hash
 
@@ -3271,6 +3271,9 @@ def get_dynamic_analysis_results(self, sample_hash=None, url=None, url_sha1=None
             else:
                 endpoint_base = self.__GET_ARCHIVE_RESULTS_ENDPOINT
 
+            hashing_algorithm = HASH_LENGTH_MAP.get(len(sample_hash))
+            endpoint_base = endpoint_base.format(hash_type=hashing_algorithm)
+
         elif url:
             if not isinstance(url, str):
                 raise WrongInputError("url parameter must be a string")
diff --git a/setup.cfg b/setup.cfg
@@ -1,2 +1,5 @@
 [metadata]
 license_files = LICENSE
+
+[aliases]
+test=pytest
diff --git a/setup.py b/setup.py
@@ -1,4 +1,5 @@
 from setuptools import setup
+from ReversingLabs.SDK import __version__
 
 
 requires = ["requests>=2.28.2", "urllib3>=1.26.14"]
@@ -11,8 +12,8 @@
 
 setup(
     name="reversinglabs-sdk-py3",
-    version="2.4.3",
-    description="Python SDK for using ReversingLabs services - Python 3 version.",
+    version=__version__,
+    description="Python SDK for using ReversingLabs services.",
     long_description=long_description,
     long_description_content_type="text/markdown",
     author="ReversingLabs",
@@ -21,6 +22,7 @@
     packages=packages,
     python_requires=">=3.6",
     install_requires=requires,
+    extras_require={"test": ["pytest"]},
     license="MIT",
     zip_safe=False,
     classifiers=[
diff --git a/tests/__init__.py b/tests/__init__.py
diff --git a/tests/test_a1000.py b/tests/test_a1000.py
@@ -0,0 +1,46 @@
+import pytest
+from ReversingLabs.SDK import __version__
+from ReversingLabs.SDK.a1000 import CLASSIFICATIONS, AVAILABLE_PLATFORMS, A1000
+from ReversingLabs.SDK.helper import WrongInputError
+
+
+EXPECTED_PLATFORMS = ("windows7", "windows10", "macos_11", "windows11", "linux")
+
+
+def test_classifications():
+	assert "macos11" not in AVAILABLE_PLATFORMS, "Did we start using macos11 in these classifications?"
+
+	assert "KNOWN" not in CLASSIFICATIONS, "Are you sure that KNOWN should be in allowed classifications?"
+	assert "GOODWARE" in CLASSIFICATIONS, "Are you sure that GOODWARE should be removed from allowed classifications?"
+
+
+def test_available_platforms():
+	for platform in AVAILABLE_PLATFORMS:
+		assert platform in EXPECTED_PLATFORMS
+
+
+def test_a1000_object():
+	invalid_host = "my.host"
+	valid_host = f"https://{invalid_host}"
+	token = "my_mock_token"
+
+	a1000 = A1000(
+		host=valid_host,
+		token=token,
+		verify=True
+	)
+
+	assert a1000._url == valid_host + "{endpoint}"
+
+	with pytest.raises(WrongInputError, match=r"host parameter must contain a protocol definition at the beginning."):
+		A1000(host=invalid_host, token=token)
+
+	with pytest.raises(WrongInputError, match=r"If token is not provided username and password are required."):
+		A1000(host=valid_host)
+
+	user_agent = a1000._headers.get("User-Agent")
+	assert __version__ in user_agent
+
+	authorization = a1000._headers.get("Authorization")
+	assert authorization == f"Token {token}"
+
diff --git a/tests/test_helper.py b/tests/test_helper.py
@@ -0,0 +1,47 @@
+from ReversingLabs.SDK.helper import *
+from ReversingLabs.SDK import __version__
+
+
+STATUS_ERRORS = (
+	NotFoundError, NoFileTypeError, UnauthorizedError, WrongInputError, ForbiddenError, BadRequestError,
+	RequestTimeoutError, ConflictError, RequestTooLargeError, InternalServerError, BadGatewayError,
+	ServiceUnavailableError, NotAllowedError, TooManyRequestsError, NotAcceptableError
+)
+OTHER_ERRORS = (
+	WrongInputError, CloudDeepScanException
+)
+ALL_ERRORS = STATUS_ERRORS + OTHER_ERRORS
+
+
+def test_vars():
+	assert HASH_LENGTH_MAP.get(32) == MD5
+	assert HASH_LENGTH_MAP.get(40) == SHA1
+	assert HASH_LENGTH_MAP.get(64) == SHA256
+	assert HASH_LENGTH_MAP.get(128) == SHA512
+
+	assert MD5 == "md5"
+	assert SHA1 == "sha1"
+	assert SHA256 == "sha256"
+	assert SHA512 == "sha512"
+
+	assert __version__ in DEFAULT_USER_AGENT
+
+
+def test_errors():
+	for err_cls in ALL_ERRORS:
+		if err_cls in OTHER_ERRORS:
+			continue
+
+		err_cls(response_object=True)
+		del err_cls
+
+
+def test_validate_hashes():
+	validate_hashes(
+		hash_input=[
+			"00f8cd09187d311707b52a1c52018e7cfb5f2f78e47bf9200f16281098741422",
+			"5377d0ed664246a604363f90a2764aa10fa63ad0",
+			"512fca9e83c47fd9c36aa7d50a856396",
+		],
+		allowed_hash_types=(SHA256, SHA1, MD5)
+	)
diff --git a/tests/test_ticloud.py b/tests/test_ticloud.py
diff --git a/tests/test_tiscale.py b/tests/test_tiscale.py
